技術的な乾物共有: HBase のデータから Kafka への移行の実践

1. 概要

実際のアプリケーション シナリオでは、データは HBase クラスターに保存されますが、何らかの特別な理由により、データを HBase から Kafka に移行する必要があります。通常の状況では、ソース データは Kafka に送信され、その後コンシューマーがデータを処理して HBase に書き込みます。しかし、逆のプロセスを実行する場合、HBase データを Kafka に移行するにはどうすればよいでしょうか?今日は具体的な実装プロセスについてお話しします。

2. コンテンツ

一般的なビジネス シナリオは次のとおりです。データ ソースがデータを生成し、Kafka に入り、その後コンシューマー (Flink、Spark、Kafka API など) によって処理されて HBase に入ります。これは典型的なリアルタイム処理フローです。フローチャートは次のとおりです。

上記のリアルタイム処理フローは、結局のところ、データフローが順次処理されるため、データの処理が比較的容易です。しかし、このプロセスを逆にすると、いくつかの問題が発生します。

2.1 膨大なデータ

HBase の分散性とクラスターの水平拡張により、HBase 内のデータは数百億、数千億、あるいはそれ以上になることがよくあります。このレベルのデータは、このタイプの逆データフローのシナリオで、データ取得の問題という非常に厄介な問題を引き起こします。この膨大な量のデータを HBase から取得するにはどうすればよいでしょうか?

2.2 データパーティションなし

HBase は高速かつ簡単にデータを取得したり一覧表示したりできることがわかっています。ただし、Hive のようなデータ ウェアハウスのパーティショニングの概念がなく、一定期間内のデータを提供できません。最新の週のデータを抽出する場合は、テーブル全体をスキャンし、タイムスタンプをフィルタリングしてその週のデータを取得する必要がある場合があります。数が少ない場合は大きな問題にならないかもしれませんが、データ量が多い場合、HBase のテーブル全体をスキャンするのは困難です。

3. 解決策

このような逆データフローをどのように処理するか。実際、これを実現するには、HBase の Get と List の機能を使用できます。 HBase は RowKey を通じてプライマリ インデックスを構築するため、RowKey レベルでのデータ取得速度は非常に高速です。実装プロセスの詳細は次のとおりです。

データフローは上の図に示されています。以下では、各プロセスの実装の詳細と注意事項を分析します。

3.1 行キーの抽出

HBase には Rowkey 取得用のプライマリ インデックスがあることがわかっているので、この機能を使用して拡張できます。 HBase テーブルから大量のデータ内の Rowkey を抽出し、設定した抽出および保存ルールに従って、抽出した Rowkey を HDFS に保存できます。

ここで注意が必要な問題の 1 つは、HBase Rowkey の抽出です。大規模なデータレベルの Rowkey 抽出の場合は、MapReduce を使用して実装することをお勧めします。これは、HBase が提供する TableMapReduceUtil クラスのおかげで実現されます。 MapReduce タスクを通じて、HBase 内の Rowkey はマップ フェーズで指定された時間範囲に従ってフィルタリングされ、reduce フェーズで Rowkey が複数のファイルに分割され、最終的に HDFS に保存されます。

ここで質問がある学生もいるかもしれません。 Rowkey の抽出には MapReduce が使用されているので、列クラスターの下の列データを直接スキャンして処理してみませんか?ここでは、MapReduce タスクを開始すると、HBase データをスキャンするときに Rowkey のみをフィルター処理し (FirstKeyOnlyFilter を使用)、列クラスター データは処理しません。これでかなり速くなります。 HBase RegionServer への負荷も大幅に軽減されます。

• 行列行001情報:名前行001情報:年齢行001情報:性別行001情報:sn

ここに例があります。上記の表のデータの場合、Rowkey (row001) のみを抽出する必要があります。ただし、実際のビジネス データでは、HBase テーブルには、多くの特徴的な属性 (名前、性別、年齢、ID カードなど) を持つデータが記述される場合があります。一部のビジネス データでは、1 つの列クラスターの下に 12 を超える特性が存在する場合がありますが、Rowkey は 1 つだけであり、必要な Rowkey はこの 1 つだけです。その場合、FirstKeyOnlyFilter を使用して実装するのが適切です。

 /**
 * フィルターは 各行の最初のKVを返します。
 * <p>
 * このフィルターを使用すると、行カウント操作をより効率的に実行できます。
 */

これは、最初の KV データを返すために使用される FirstKeyOnlyFilter の機能の説明です。実際に役人が数えるのに使っています。ここで少し改善して、FirstKeyOnlyFilter を使用して Rowkey を抽出します。

3.2 行キー生成

抽出された Rowkey を生成するにはどうすればいいですか? Reduce の数は、実際の桁数に基づいて決定できます。 Rowkey ファイルを生成するときは、実際のデータ量に基づいて Reduce の数を計算することをお勧めします。使いやすさを優先して、HDFS ファイルを 1 つだけ使用しないようにしてください。後でメンテナンスが難しくなります。たとえば、HBase テーブルが 100 GB の場合、それを 100 個のファイルに分割できます。

3.3 データ処理

ステップ 1 では、抽出ルールと保存ルールに従って、MapReduce を介して HBase からデータが抽出され、RowKey が HDFS に保存されます。次に、MapReduce タスクを通じて HDFS 上の Rowkey ファイルを読み取り、List を通じて HBase からデータを取得します。分解の詳細は次のとおりです。

Map フェーズでは、HDFS から Rowkey データ ファイルを読み取り、バッチ Get を通じて HBase からデータを取得し、データを組み立てて Reduce フェーズに送信します。 Reduce フェーズでは、Map フェーズからのデータが取得され、Kafka に書き込まれます。 Kafka プロデューサー コールバック関数は、Kafka に書き込まれるデータのステータス情報を取得するために使用され、ステータス情報はデータが正常に書き込まれたかどうかを判断するために使用されます。成功した場合は、成功した進行状況の統計を容易にするために、成功した Rowkey を HDFS に記録します。失敗した場合は、失敗した進行状況の統計を容易にするために、失敗した Rowkey を HDFS に記録します。

3.4 失敗と再試行

MapReduce タスクを通じて Kafka にデータを書き込むときに、失敗が発生する可能性があります。障害が発生した場合は、HDFS に Rowkey を記録するだけで済みます。タスクが完了すると、プログラムは HDFS 上に失敗した Rowkey ファイルがあるかどうかを確認します。存在する場合は、手順 3 を再度開始します。つまり、HDFS 上の失敗した Rowkey ファイルを読み取り、HBase にデータをリストし、データを処理し、最後に Kafka に書き込み、HDFS 上の失敗した Rowkey が処理されるまでこれを繰り返します。

4. 実装コード

ここで実装されているコードの量は複雑ではありません。以下は、これに基づいて変更できる疑似コードです (たとえば、Rowkey を抽出し、MapReduce を使用して Rowkey を読み取り、HBase テーブルをバッチで取得してから、Kafka に書き込むなど)。サンプルコードは次のとおりです。

パブリッククラスMRROW2HDFS {
公共 静的void main(String[] args)は例外をスローします{
 構成 config = HBaseConfiguration。作成する（）; // HBase 構成情報
 ジョブ job = Job.getInstance(config, "MRROW2HDFS" );
 ジョブにJarByClassを設定します(MRROW2HDFS.class);
 ROWReducer クラスをジョブに設定します。
 文字列 hbaseTableName = "hbase_tbl_name" ;
 スキャン scan = new Scan();
 スキャンキャッシュの設定(1000);
 スキャン.setCacheBlocks( false );
 scan.setFilter(新しい FirstKeyOnlyFilter());
 TableMapReduceUtil.initTableMapperJob(hbaseTableName、スキャン、ROWMapper.class、Text.class、Text.class、ジョブ);
 FileOutputFormat.setOutputPath(ジョブ、新しいパス( "/tmp/rowkey.list" )); // ストレージの行キーの HDFs パスを入力します
 System.exit(job.waitForCompletion( true ) ? 0 : 1);
 }
公共 静的クラス ROWMapper は TableMapper<Text, Text> を拡張します {
 @オーバーライド
 protected void map(ImmutableBytesWritableキー, 結果値,
 Mapper<ImmutableBytesWritable, Result, Text, Text>.Context コンテキスト)
 IOException、InterruptedException をスローします {
 for (セル cell : value.rawCells()) {
 //日付範囲をフィルターする
 // コンテキストを書き込みます(...);
 }
 }
 } 
  
公共 静的クラス ROWReducer は Reducer<Text,Text,Text,Text> を拡張します{
 プライベートテキスト結果 = 新しいテキスト(); 
  
 @オーバーライド
 protected void Reduce(Text key , Iterable<Text> values ,Context context) は IOException、InterruptedException をスローします {
 for (テキスト値:値​​){
 結果を設定します(val);
 context.write(キー, 結果 );
 }
 }
 }
 }

5. まとめ

逆データ処理プロセス全体は複雑ではなく、実装は論理処理が複雑すぎない非常に基本的な MapReduce ロジックです。処理中は、いくつかの詳細を考慮する必要があります。 Rowkey が HDFS 上で生成される場合、行内にスペースが含まれることがあります。 HDFS 上の Rowkey ファイルをリストに読み込むときは、データごとにスペースをフィルタリングするのが最適です。さらに、失敗したタスクの再実行とデータの調整を容易にするために、成功した Rowkey 処理と失敗した Rowkey 処理の記録が保持されます。データ移行の進行状況と完了状況を知ることができます。同時に、Kafka Eagle 監視ツールを使用して、Kafka の書き込みの進行状況を表示することもできます。