本文為您介紹如何同步EMR DataFlow集群的數據至EMR DataLake集群的HDFS。

前提條件

  • 已創建DataLake集群,并且選擇了Flume服務,詳情請參見創建集群
  • 已創建DataFlow集群,并且選擇了Kafka服務,詳情請參見創建集群

操作步驟

  1. 配置Flume。
    1. 進入Flume的配置頁面。
      1. 登錄EMR on ECS控制臺
      2. 在頂部菜單欄處,根據實際情況選擇地域和資源組
      3. 集群管理頁面,單擊目標集群操作列的集群服務
      4. 集群服務頁面,單擊FLUME服務區域的配置
    2. 單擊flume-conf.properties頁簽。
      本文示例采用的是全局配置方式,如果您想按照節點配置,可以在FLUME服務配置頁面的下拉列表中選擇獨立節點配置
    3. 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
      
      # Describe the sink
      default-agent.sinks.k1.type = hdfs
      default-agent.sinks.k1.hdfs.path = hdfs://emr-cluster/tmp/flume/test-data
      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 DataFlow集群Broker的Host和端口號。
      default-agent.sources.source1.kafka.topics Flume消費Kafka數據的Topic。
      default-agent.channels.c1.capacity 通道中存儲的最大事件數。請根據實際環境修改該參數值。
      default-agent.channels.c1.transactionCapacity 每個事務通道將從源接收或提供給接收器的最大事件數。請根據實際環境修改該參數值。
      default-agent.sinks.k1.hdfs.path Flume向HDFS寫入數據的路徑。
      針對普通集群和高可用集群代碼示例如下:
      • 高可用集群
        default-agent.sinks.k1.hdfs.path = hdfs://emr-cluster/tmp/flume/test-data
      • 普通集群
        default-agent.sinks.k1.hdfs.path = hdfs://master-1-1:9000/tmp/flume/test-data
    4. 保存配置。
      1. 單擊下方的保存
      2. 在彈出的對話框中,輸入執行原因,單擊確定
  2. 啟動服務。
    1. 在FLUME服務頁面,選擇更多操作 > 重啟
    2. 在彈出的對話框中,輸入執行原因,單擊確定
    3. 確認對話框中,單擊確定
  3. 測試數據同步情況。
    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并回車。

    4. 通過SSH方式連接DataLake集群,查看生成的文件。
      Flume會在HDFS中以當前時間的時間戳(毫秒)生成文件FlumeData.xxxx。文件內容是在Kafka中輸入的數據abc
      hdfs dfs -cat /tmp/flume/test-data/<FlumeData.xxxx>