How to Migrate Data from HBase to Kafka: A Reverse Data Flow Guide

In typical real‑time pipelines data flows from the source into Kafka, then consumers such as Flink, Spark, or the Kafka API process the data and store it in HBase. Occasionally a reverse flow is required, moving data from HBase back into Kafka, which introduces two major challenges: massive data volume and the absence of partitioning in HBase.

Because HBase provides a primary index on RowKey, the author proposes extracting only the RowKeys. A MapReduce job scans the HBase table with FirstKeyOnlyFilter, writes the extracted RowKeys to HDFS, and avoids full table scans of column families.

During the Map phase, the job reads the RowKey files from HDFS, performs bulk Get operations on HBase, and emits the retrieved rows to the Reduce phase. The Reduce phase writes each record to Kafka via a producer, records the status of each write, and stores successful and failed RowKeys back to HDFS for later analysis.

Failed writes are handled by re‑running the MapReduce job on the failure file until all rows are successfully written. Progress can be monitored with tools such as Kafka Eagle.

The article provides a concise pseudo‑code implementation that includes the driver class, a mapper that configures the Scan with FirstKeyOnlyFilter, and a reducer that writes to Kafka. The code uses HBaseConfiguration, TableMapReduceUtil, and standard Hadoop APIs.

Overall, the reverse data migration process is straightforward but requires careful handling of whitespace in RowKey files, appropriate partitioning of output files based on data size, and diligent logging of success and failure to enable reliable retries.

public class MRROW2HDFS {
    public static void main(String[] args) throws Exception {
        Configuration config = HBaseConfiguration.create(); // HBase Config info
        Job job = Job.getInstance(config, "MRROW2HDFS");
        job.setJarByClass(MRROW2HDFS.class);
        job.setReducerClass(ROWReducer.class);
        String hbaseTableName = "hbase_tbl_name";
        Scan scan = new Scan();
        scan.setCaching(1000);
        scan.setCacheBlocks(false);
        scan.setFilter(new FirstKeyOnlyFilter());
        TableMapReduceUtil.initTableMapperJob(hbaseTableName, scan, ROWMapper.class, Text.class, Text.class, job);
        FileOutputFormat.setOutputPath(job, new Path("/tmp/rowkey.list")); // input you storage rowkey hdfs path
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }

    public static class ROWMapper extends TableMapper<Text, Text> {
        @Override
        protected void map(ImmutableBytesWritable key, Result value,
                Mapper<ImmutableBytesWritable, Result, Text, Text>.Context context)
                throws IOException, InterruptedException {
            for (Cell cell : value.rawCells()) {
                // Filter date range
                // context.write(...);
            }
        }
    }

    public static class ROWReducer extends Reducer<Text, Text, Text, Text> {
        private Text result = new Text();
        @Override
        protected void reduce(Text key, Iterable<Text> values, Context context)
                throws IOException, InterruptedException {
            for (Text val : values) {
                result.set(val);
                context.write(key, result);
            }
        }
    }
}