Object Storage Service:Alirkan data EMR Kafka ke OSS-HDFS dengan Flume

Prosedur

1. Konfigurasikan Flume.

   1. Buka halaman konfigurasi Flume.

      1. Masuk ke Konsol E-MapReduce.

      2. Di bilah navigasi atas, pilih Wilayah dan kelompok sumber daya.

      3. Di halaman EMR on ECS, temukan kluster target lalu klik Services di kolom Actions.

      4. Di tab Services, klik Configure di area layanan FLUME.

   2. Atur ukuran maksimum heap Java Virtual Machine (JVM) (Xmx).

      Menulis data dari Flume ke layanan OSS-HDFS membutuhkan banyak memori. Kami menyarankan Anda menambah nilai Xmx untuk Agen Flume.

      1. Klik tab flume-env.sh.

         Topik ini menggunakan konfigurasi global. Untuk mengonfigurasi node tertentu, pilih Independent Node Configuration dari daftar drop-down di halaman Configure untuk layanan FLUME.

      2. Ubah nilai parameter JAVA_OPTS.

         Sebagai contoh, untuk mengatur ukuran maksimum heap JVM menjadi 1 GB, ubah nilai parameter menjadi -Xmx1g.

      3. Klik Save.

   3. Ubah konfigurasi flume-conf.properties.

      1. Klik tab flume-conf.properties.

         Topik ini menggunakan konfigurasi global. Untuk mengonfigurasi node tertentu, pilih Independent Node Configuration dari daftar drop-down di halaman Configure untuk layanan FLUME.

      2. Di editor untuk flume-conf.properties, masukkan konfigurasi berikut.

         Catatan

         Nilai default-agent pada contoh berikut harus sesuai dengan nilai parameter agent_name di halaman Configure untuk layanan FLUME.

         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
         
         # Gunakan saluran memori untuk menyimpan sementara event
         default-agent.channels.c1.type = memory
         default-agent.channels.c1.capacity = 100
         default-agent.channels.c1.transactionCapacity = 100
         
         # Hubungkan source dan sink ke saluran
         default-agent.sources.source1.channels = c1
         default-agent.sinks.k1.channel = c1

         Parameter	Deskripsi
         default-agent.sources.source1.kafka.bootstrap.servers	Hostname dan nomor port broker di kluster Kafka Anda.
         default-agent.sinks.k1.hdfs.path	Jalur untuk OSS-HDFS harus dalam format oss://{yourBucketName}.{yourBucketRegion}.oss-dls.aliyuncs.com/{path}. Contohnya: oss://flume-test.cn-hangzhou.oss-dls.aliyuncs.com/result. Komponen jalur: {yourBucketName}: Masukkan nama Bucket tempat layanan OSS-HDFS diaktifkan. {yourBucketRegion}: Masukkan ID wilayah Bucket tersebut. {path}: Masukkan nama direktori layanan OSS-HDFS.
         default-agent.channels.c1.capacity	Jumlah maksimum event yang dapat disimpan di saluran. Sesuaikan nilai ini berdasarkan lingkungan Anda.
         default-agent.channels.c1.transactionCapacity	Jumlah maksimum event yang dapat ditulis oleh source atau diambil oleh sink dari saluran per transaksi. Sesuaikan nilai ini berdasarkan lingkungan Anda.

      3. Klik Save.

2. Uji sinkronisasi data.

   1. Sambungkan ke kluster EMR Dataflow menggunakan SSH. Untuk informasi selengkapnya, lihat Masuk ke kluster.
   2. Buat topik bernama flume-test.

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

   3. Hasilkan data uji.

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

      Sebagai contoh, masukkan abc lalu tekan Enter.

      Sebuah file akan dibuat di jalur oss://flume-test.cn-hangzhou.oss-dls.aliyuncs.com/result. File tersebut diberi nama dalam format FlumeData.xxxx, dengan stempel waktu saat ini dalam milidetik sebagai akhiran.