Object Storage Service:Stream EMR Kafka data to OSS-HDFS with Flume

Procedure

1. Configure Flume.

   1. Go to the Flume configuration page.

      1. Log on to the E-MapReduce console.

      2. In the top navigation bar, select a region and resource group.

      3. On the EMR on ECS page, find the target cluster and click Services in the Actions column.

      4. On the Services tab, click Configure in the FLUME service area.

   2. Set the maximum Java Virtual Machine (JVM) heap size (Xmx).

      Writing data from Flume to the OSS-HDFS service is memory-intensive. We recommend that you increase the Xmx value for the Flume agent.

      1. Click the flume-env.sh tab.

         This topic uses global configuration. To configure a specific node, select Independent Node Configuration from the drop-down list on the Configure page for the FLUME service.

      2. Modify the value of the JAVA_OPTS parameter.

         For example, to set the maximum JVM heap size to 1 GB, change the parameter value to -Xmx1g.

      3. Click Save.

   3. Modify the flume-conf.properties configuration.

      1. Click the flume-conf.properties tab.

         This topic uses global configuration. To configure a specific node, select Independent Node Configuration from the drop-down list on the Configure page for the FLUME service.

      2. In the editor for flume-conf.properties, enter the following configuration.

         Note

         The value of default-agent in the following example must match the value of the agent_name parameter on the Configure page for the FLUME service.

         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

         Parameter	Description
         default-agent.sources.source1.kafka.bootstrap.servers	The hostnames and port numbers of the brokers in your Kafka cluster.
         default-agent.sinks.k1.hdfs.path	The path for OSS-HDFS must be in the format oss://{yourBucketName}.{yourBucketRegion}.oss-dls.aliyuncs.com/{path}. For example: oss://flume-test.cn-hangzhou.oss-dls.aliyuncs.com/result. Path components: {yourBucketName}: Enter the name of the Bucket for which the OSS-HDFS service is enabled. {yourBucketRegion}: Enter the region ID of the Bucket. {path}: Enter the directory name of the OSS-HDFS service.
         default-agent.channels.c1.capacity	The maximum number of events that can be stored in the channel. Adjust this value based on your environment.
         default-agent.channels.c1.transactionCapacity	The maximum number of events that a source can write to or a sink can retrieve from the channel per transaction. Adjust this value based on your environment.

      3. Click Save.

2. Test the data synchronization.

   1. Connect to the EMR Dataflow cluster using SSH. For more information, see Log on to a cluster.
   2. Create a topic named flume-test.

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

   3. Produce test data.

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

      For example, enter abc and press Enter.

      A file is generated in the oss://flume-test.cn-hangzhou.oss-dls.aliyuncs.com/result path. The file is named in the FlumeData.xxxx format, with the current timestamp in milliseconds as the suffix.