Object Storage Service:Flume を使用した EMR Kafka データの OSS-HDFS へのストリーミング

操作手順

1. Flume の構成を行います。

   1. Flume 構成ページに移動します。

      1. E-MapReduce コンソールにログインします。

      2. 上部ナビゲーションバーでリージョン およびリソースグループを選択します。

      3. EMR on ECS ページで対象のクラスターを見つけ、**[操作]** 列の Services をクリックします。

      4. Services タブで、FLUME サービス領域の Configure をクリックします。

   2. Java 仮想マシン (JVM) ヒープサイズの最大値 (Xmx) を設定します。

      Flume から OSS-HDFS サービスへのデータ書き込みはメモリを多く消費します。Flume エージェントの Xmx 値を増やすことを推奨します。

      1. flume-env.sh タブをクリックします。

         本ドキュメントではグローバル構成を使用しています。特定のノードを構成する場合は、FLUME サービスの Configure ページにあるドロップダウンリストから Independent Node Configuration を選択してください。

      2. JAVA_OPTS パラメーターの値を変更します。

         たとえば、JVM ヒープサイズの最大値を 1 GB に設定する場合、パラメーター値を -Xmx1g に変更します。

      3. 保存をクリックします。

   3. flume-conf.properties 構成を変更します。

      1. flume-conf.properties タブをクリックします。

         本ドキュメントではグローバル構成を使用しています。特定のノードを構成する場合は、FLUME サービスの Configure ページにあるドロップダウンリストから Independent Node Configuration を選択してください。

      2. flume-conf.properties のエディターに、次の構成を入力します。

         説明

         以下の例における default-agent の値は、FLUME サービスの Configure ページ上の agent_name パラメーターの値と一致させる必要があります。

         default-agent.sources = source1
         default-agent.sinks = k1
         default-agent.channels = c1
         
         default-agent.sources.source1.type = org.apache.flume.source.kafka.KafkaSource
         default-agent.sources.source1.channels = c1
         default-agent.sources.source1.kafka.bootstrap.servers = <kafka-host1:port1,kafka-host2:port2...>
         default-agent.sources.source1.kafka.topics = flume-test
         default-agent.sources.source1.kafka.consumer.group.id = flume-test-group
         
         default-agent.sinks.k1.type = hdfs
         default-agent.sinks.k1.hdfs.path = oss://{yourBucketName}.{yourBucketRegion}.oss-dls.aliyuncs.com/{path}
         default-agent.sinks.k1.hdfs.fileType=DataStream
         
         # Use a memory channel to buffer events
         default-agent.channels.c1.type = memory
         default-agent.channels.c1.capacity = 100
         default-agent.channels.c1.transactionCapacity = 100
         
         # Bind the source and sink to the channel
         default-agent.sources.source1.channels = c1
         default-agent.sinks.k1.channel = c1

         パラメーター	説明
         default-agent.sources.source1.kafka.bootstrap.servers	Kafka クラスター内のブローカーのホスト名とポート番号。
         default-agent.sinks.k1.hdfs.path	OSS-HDFS のパスは、oss://{yourBucketName}.{yourBucketRegion}.oss-dls.aliyuncs.com/{path} 形式である必要があります。例：oss://flume-test.cn-hangzhou.oss-dls.aliyuncs.com/result。 パスの構成要素： {yourBucketName}：OSS-HDFS サービスが有効化されているバケット名を入力します。 {yourBucketRegion}：バケットのリージョン ID を入力します。 {path}：OSS-HDFS サービスのディレクトリ名を入力します。
         default-agent.channels.c1.capacity	チャンネルに格納可能なイベントの最大数。環境に応じてこの値を調整してください。
         default-agent.channels.c1.transactionCapacity	1 トランザクションあたり、ソースがチャンネルに書き込める、またはシンクがチャンネルから取得できるイベントの最大数。環境に応じてこの値を調整してください。

      3. 保存をクリックします。

2. データ同期をテストします。

   1. SSH を使用して EMR Dataflow クラスターに接続します。詳細については、「クラスターへのログイン」をご参照ください。
   2. flume-test という名前のトピックを作成します。

      kafka-topics.sh --partitions 10 --replication-factor 2 --zookeeper master-1-1:2181/emr-kafka --topic flume-test --create

   3. テストデータを生成します。

      kafka-console-producer.sh --topic flume-test --broker-list master-1-1:9092

      たとえば、abc と入力し、Enter キーを押します。

      oss://flume-test.cn-hangzhou.oss-dls.aliyuncs.com/result パスにファイルが生成されます。ファイル名は FlumeData.xxxx 形式で、サフィックスには現在のタイムスタンプ（ミリ秒）が付与されます。