public static void writeIndex(final FileSystem fileSystem, final Path sstablePath) throws IOException {

        final Configuration configuration = fileSystem.getConf();

        final long splitSize = configuration.getLong(HadoopSSTableConstants.HADOOP_SSTABLE_SPLIT_MB,
                HadoopSSTableConstants.DEFAULT_SPLIT_MB) * 1024 * 1024;

        final Closer closer = Closer.create();

        final Path outputPath = sstablePath.suffix(SSTABLE_INDEX_SUFFIX);
        final Path inProgressOutputPath = sstablePath.suffix(SSTABLE_INDEX_IN_PROGRESS_SUFFIX);

        boolean success = false;
        try {
            final FSDataOutputStream os = closer.register(fileSystem.create(inProgressOutputPath));

            final TLongArrayList splitOffsets = new TLongArrayList();
            long currentStart = 0;
            long currentEnd = 0;
            final IndexOffsetScanner index = closer.register(new IndexOffsetScanner(sstablePath, fileSystem));

            while (index.hasNext()) {
                // NOTE: This does not give an exact size of this split in bytes but a rough estimate.
                // This should be good enough since it's only used for sorting splits by size in hadoop land.
                while (currentEnd - currentStart < splitSize && index.hasNext()) {
                    currentEnd = index.next();
                    splitOffsets.add(currentEnd);
                }

                // Record the split
                final long[] offsets = splitOffsets.toArray();
                os.writeLong(offsets[0]); // Start
                os.writeLong(offsets[offsets.length - 1]); // End

                // Clear the offsets
                splitOffsets.clear();

                if (index.hasNext()) {
                    currentStart = index.next();
                    currentEnd = currentStart;
                    splitOffsets.add(currentStart);
                }
            }

            success = true;
        } finally {
            closer.close();

            if (!success) {
                fileSystem.delete(inProgressOutputPath, false);
            } else {
                fileSystem.rename(inProgressOutputPath, outputPath);
            }
        }
    }
public static void writeIndex ( final FileSystem fileSystem , final Path sstablePath ) throws IOException { final Configuration configuration = fileSystem . getConf ( ) ; final long splitSize = configuration . getLong ( HadoopSSTableConstants . HADOOP_SSTABLE_SPLIT_MB , HadoopSSTableConstants . DEFAULT_SPLIT_MB ) * 1024 * 1024 ; final Closer closer = Closer . create ( ) ; final Path outputPath = sstablePath . suffix ( SSTABLE_INDEX_SUFFIX ) ; final Path inProgressOutputPath = sstablePath . suffix ( SSTABLE_INDEX_IN_PROGRESS_SUFFIX ) ; boolean success = false ; try { final FSDataOutputStream os = closer . register ( fileSystem . create ( inProgressOutputPath ) ) ; final TLongArrayList splitOffsets = new TLongArrayList ( ) ; long currentStart = 0 ; long currentEnd = 0 ; final IndexOffsetScanner index = closer . register ( new IndexOffsetScanner ( sstablePath , fileSystem ) ) ; while ( index . hasNext ( ) ) { while ( currentEnd - currentStart < splitSize && index . hasNext ( ) ) { currentEnd = index . next ( ) ; splitOffsets . add ( currentEnd ) ; } final long [ ] offsets = splitOffsets . toArray ( ) ; os . writeLong ( offsets [ 0 ] ) ; os . writeLong ( offsets [ offsets . length - 1 ] ) ; splitOffsets . clear ( ) ; if ( index . hasNext ( ) ) { currentStart = index . next ( ) ; currentEnd = currentStart ; splitOffsets . add ( currentStart ) ; } } success = true ; } finally { closer . close ( ) ; if ( ! success ) { fileSystem . delete ( inProgressOutputPath , false ) ; } else { fileSystem . rename ( inProgressOutputPath , outputPath ) ; } } }
Create and write an index index based on the input Cassandra Index.db file. Read the Index.db and generate chunks
(splits) based on the configured chunk size.

@param fileSystem Hadoop file system.
@param sstablePath SSTable Index.db.
@throws IOException