使用Flume同步EMR Kafka集群的数据至OSS-HDFS服务

操作步骤

1. 配置Flume。

   1. 进入Flume的配置页面。

      1. 登录E-MapReduce控制台。

      2. 在顶部菜单栏处，根据实际情况选择地域和资源组。

      3. 在EMR on ECS页面，单击目标集群操作列的集群服务。

      4. 在集群服务页签，单击FLUME服务区域的配置。

   2. 设置JVM最大可用内存（Xmx）。

      Flume向OSS-HDFS写入数据时需要占用较大的JVM内存，建议增加Flume Agent的Xmx。具体步骤如下：

      1. 单击flume-env.sh页签。

         本文采用了全局配置方式。如果您希望按照节点配置，可以在FLUME服务配置页面的下拉列表中选择独立节点配置。

      2. 修改JAVA_OPTS的参数值。

         例如，JVM最大可用内存设置为1 GB，则参数值修改为-Xmx1g。

      3. 单击保存。

   3. 修改flume-conf.properties配置。

      1. 单击flume-conf.properties页签。

         本文采用了全局配置方式。如果您希望按照节点配置，可以在FLUME服务配置页面的下拉列表中选择独立节点配置。

      2. 在flume-conf.properties右侧，输入以下配置项。

         说明

         以下示例中的default-agent的值需与FLUME服务配置页面的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://<examplebucket>.<exampleregion>.oss-dls.aliyuncs.com/<exampledir>
         default-agent.sinks.k1.hdfs.fileType=DataStream
         
         # Use a channel which buffers events in memory
         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集群Broker的Host和端口号。
         default-agent.sinks.k1.hdfs.path	OSS-HDFS的路径。填写格式为oss://<examplebucket>.<exampleregion>.oss-dls.aliyuncs.com/<exampledir>。示例值为oss://flume-test.cn-hangzhou.oss-dls.aliyuncs.com/result。 各参数说明如下： <examplebucket>：填写已开启OSS-HDFS服务的Bucket名称。 <exampleregion>：填写Bucket所在的地域ID。 <exampledir>：填写OSS-HDFS服务的目录名称。
         default-agent.channels.c1.capacity	通道中存储的最大事件数。请根据实际环境修改该参数值。
         default-agent.channels.c1.transactionCapacity	每个事务通道将从源接收或提供给接收器的最大事件数。请根据实际环境修改该参数值。

      3. 单击保存。

2. 测试数据同步情况。

   1. 通过SSH方式连接DataFlow集群，详情请参见登录集群。
   2. 创建名称为flume-test的Topic。

      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并回车。

      在oss://flume-test.cn-hangzhou.oss-dls.aliyuncs.com/result路径下会以当前时间的时间戳（毫秒）为后缀生成格式为FlumeData.xxxx的文件。