Object Storage Service:Flume ile EMR Kafka verilerini OSS-HDFS'e aktarın

Önkoşullar

• OSS-HDFS hizmeti etkinleştirilmiş ve gerekli erişim izinlerine sahipsiniz. Daha fazla bilgi için bkz. OSS-HDFS hizmetini etkinleştirme.

• Bir DataLake kümesi oluşturdunuz ve Flume hizmetini seçtiniz. Daha fazla bilgi için bkz. Küme oluşturma.

• Bir DataFlow kümesi oluşturdunuz ve Kafka hizmetini seçtiniz. Daha fazla bilgi için bkz. Küme oluşturma.

Yordam

1. Flume'u yapılandırın.

   1. Flume yapılandırma sayfasına gidin.

      1. E-MapReduce konsolu üzerinden oturum açın.

      2. Üst gezinme çubuğunda bir bölge ve kaynak grubu seçin.

      3. EMR on ECS sayfasında hedef kümeyi bulun ve İşlemler sütunundaki Services seçeneğine tıklayın.

      4. Services sekmesinde, FLUME hizmet alanındaki Configure seçeneğine tıklayın.

   2. Maksimum Java Sanal Makinesi (JVM) yığın boyutunu (Xmx) ayarlayın.

      Flume'dan OSS-HDFS hizmetine veri yazmak yoğun bellek kullanımı gerektirir. Flume aracı için Xmx değerini artırmanızı öneririz.

      1. flume-env.sh sekmesine tıklayın.

         Bu konuda genel yapılandırma kullanılmaktadır. Belirli bir düğümü yapılandırmak için FLUME hizmetinin Configure sayfasındaki açılır listeden Independent Node Configuration seçeneğini belirleyin.

      2. JAVA_OPTS parametresinin değerini değiştirin.

         Örneğin, maksimum JVM yığın boyutunu 1 GB olarak ayarlamak için parametre değerini -Xmx1g olarak değiştirin.

      3. Kaydedin.

   3. flume-conf.properties yapılandırmasını değiştirin.

      1. flume-conf.properties sekmesine tıklayın.

         Bu konuda genel yapılandırma kullanılmaktadır. Belirli bir düğümü yapılandırmak için FLUME hizmetinin Configure sayfasındaki açılır listeden Independent Node Configuration seçeneğini belirleyin.

      2. flume-conf.properties düzenleyicisine aşağıdaki yapılandırmayı girin.

         Not

         Aşağıdaki örnekteki default-agent değeri, FLUME hizmetinin Configure sayfasındaki agent_name parametresinin değeriyle eşleşmelidir.

         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

         Parametre	Açıklama
         default-agent.sources.source1.kafka.bootstrap.servers	Kafka kümenizdeki broker'ların ana bilgisayar adları ve bağlantı noktası numaraları.
         default-agent.sinks.k1.hdfs.path	OSS-HDFS yolu oss://{yourBucketName}.{yourBucketRegion}.oss-dls.aliyuncs.com/{path} biçiminde olmalıdır. Örneğin: oss://flume-test.cn-hangzhou.oss-dls.aliyuncs.com/result. Yol bileşenleri: {yourBucketName}: OSS-HDFS hizmetinin etkinleştirildiği paketin adını girin. {yourBucketRegion}: Paketin bölge kimliğini girin. {path}: OSS-HDFS hizmetinin dizin adını girin.
         default-agent.channels.c1.capacity	Kanalda depolanabilecek maksimum olay sayısı. Bu değeri ortamınıza göre ayarlayın.
         default-agent.channels.c1.transactionCapacity	Bir işlem başına bir kaynağın kanala yazabileceği veya bir havuzun kanaldan alabileceği maksimum olay sayısı. Bu değeri ortamınıza göre ayarlayın.

      3. Kaydedin.

2. Veri senkronizasyonunu test edin.

   1. SSH kullanarak EMR Dataflow kümesine bağlanın. Daha fazla bilgi için bkz. Kümeye oturum açma.

   2. flume-test adlı bir konu oluşturun.

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

   3. Test verisi üretin.

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

      Örneğin, abc girin ve Enter tuşuna basın.

      oss://flume-test.cn-hangzhou.oss-dls.aliyuncs.com/result yolunda bir dosya oluşturulur. Dosya adı FlumeData.xxxx biçimindedir ve sonek olarak milisaniye cinsinden geçerli zaman damgasını içerir.