Flume 配置说明

准备工作

• 已登录阿里云 E-MapReduce 控制台。
• 创建Hadoop集群时，在可选服务中选择Flume。
• 创建Kafka集群，并创建名称为flume-test的topic，用于生成数据。

Kafka->HDFS

• 配置Flume
  创建配置文件flume.properties，添加如下配置。其中，配置项a1.sources.source1.kafka.bootstrap.servers为Kafka集群broker的host和端口号，a1.sources.source1.kafka.topics为Flume消费Kafka数据的topic，a1.sinks.k1.hdfs.path为Flume向HDFS写入数据的路径。

  a1.sources = source1
  a1.sinks = k1
  a1.channels = c1
  
  a1.sources.source1.type = org.apache.flume.source.kafka.KafkaSource
  a1.sources.source1.channels = c1
  a1.sources.source1.kafka.bootstrap.servers = kafka-host1:port1,kafka-host2:port2...
  a1.sources.source1.kafka.topics = flume-test
  a1.sources.source1.kafka.consumer.group.id = flume-test-group
  
  # Describe the sink
  a1.sinks.k1.type = hdfs
  a1.sinks.k1.hdfs.path = /tmp/flume/test-data
  a1.sinks.k1.hdfs.fileType=DataStream
  
  # Use a channel which buffers events in memory
  a1.channels.c1.type = memory
  a1.channels.c1.capacity = 100
  a1.channels.c1.transactionCapacity = 100
  
  # Bind the source and sink to the channel
  a1.sources.source1.channels = c1
  a1.sinks.k1.channel = c1

  说明 配置项a1.sinks.k1.hdfs.path如果使用URL的形式，对于高可用集群，使用hdfs://emr-cluster，如下所示。

  a1.sinks.k1.hdfs.path = hdfs://emr-cluster/tmp/flume/test-data

  对于标准集群，使用hdfs://emr-header-1:9000，如下所示。

  a1.sinks.k1.hdfs.path = hdfs://emr-header-1:9000/tmp/flume/test-data

• 启动服务
  Flume的默认配置文件放在/etc/ecm/flume-conf下，使用该配置启动Flume Agent。

  flume-ng agent --name a1 --conf /etc/ecm/flume-conf  --conf-file flume.properties

  启动Agent后，因为使用了/etc/ecm/flume-conf下的log4j.properties，会在当前路径下生成日志logs/flume.log，可根据实际使用对log4j.properties进行配置。

• 测试
  在Kafka集群使用kafka-console-producer.sh输入测试数据abc。
  Flume会在HDFS中以当前时间的（毫秒）时间戳生成文件FlumeData.xxxx，查看文件内容，会看到在Kafka中输入的数据。

Kafka->Hive

• 创建Hive表
  Flume使用事务操作将数据写入Hive，需要在创建Hive表时设置 transactional属性，如创建flume_test表。

  create table flume_test (id int, content string)
  clustered by (id) into 2 buckets
  stored as orc  TBLPROPERTIES('transactional'='true');

• 配置Flume
  创建配置文件flume.properties，添加如下配置。其中，配置项a1.sources.source1.kafka.bootstrap.servers填写Kafka集群broker的host和端口号，a1.sinks.k1.hive.metastore为Hive metastore URI，配置为hive-site.xml中配置项 hive.metastore.uris的值。

  a1.sources = source1
  a1.sinks = k1
  a1.channels = c1
  
  a1.sources.source1.type = org.apache.flume.source.kafka.KafkaSource
  a1.sources.source1.channels = c1
  a1.sources.source1.kafka.bootstrap.servers = kafka-host1:port1,kafka-host2:port2...
  a1.sources.source1.kafka.topics = flume-test
  a1.sources.source1.kafka.consumer.group.id = flume-test-group
  
  # Describe the sink
  a1.sinks.k1.type = hive
  a1.sinks.k1.hive.metastore = thrift://xxxx:9083
  a1.sinks.k1.hive.database = default
  a1.sinks.k1.hive.table = flume_test
  a1.sinks.k1.serializer = DELIMITED
  a1.sinks.k1.serializer.delimiter = ","
  a1.sinks.k1.serializer.serdeSeparator = ','
  a1.sinks.k1.serializer.fieldnames =id,content
  
  a1.channels.c1.type = memory
  a1.channels.c1.capacity = 100
  a1.channels.c1.transactionCapacity = 100
  
  a1.sources.source1.channels = c1
  a1.sinks.k1.channel = c1

• 启动Flume

  flume-ng agent --name a1 --conf /etc/ecm/flume-conf  --conf-file flume.properties

• 生成数据

  在Kafka集群中使用kafka-console-producer.sh，以逗号为分隔符输入测试数据 1，a。

• 检测数据写入
  查询Hive事务表需要在客户端进行配置。

  hive.support.concurrency – true
  hive.exec.dynamic.partition.mode – nonstrict
  hive.txn.manager – org.apache.hadoop.hive.ql.lockmgr.DbTxnManager

  配置好后查询flume_test表中的数据。

  

Kafka->HBase

• 创建HBase表
  创建HBase表flume_test及列簇column。
• 配置Flume
  创建配置文件flume.properties，添加如下配置。其中，配置项 a1.sources.source1.kafka.bootstrap.servers 为Kafka集群 broker的host和端口号，a1.sinks.k1.table 为HBase表名，a1.sinks.k1.columnFamily 为列簇名：

  a1.sources = source1
  a1.sinks = k1
  a1.channels = c1
  
  a1.sources.source1.type = org.apache.flume.source.kafka.KafkaSource
  a1.sources.source1.channels = c1
  a1.sources.source1.kafka.bootstrap.servers = kafka-host1:port1,kafka-host2:port2...
  a1.sources.source1.kafka.topics = flume-test
  a1.sources.source1.kafka.consumer.group.id = flume-test-group
  
  a1.sinks.k1.type = hbase
  a1.sinks.k1.table = flume_test
  a1.sinks.k1.columnFamily = column
  
  
  # Use a channel which buffers events in memory
  a1.channels.c1.type = memory
  a1.channels.c1.capacity = 1000
  a1.channels.c1.transactionCapacity = 100
  
  # Bind the source and sink to the channel
  a1.sources.source1.channels = c1
  a1.sinks.k1.channel = c1

• 启动服务

  flume-ng agent --name a1 --conf /etc/ecm/flume-conf  --conf-file flume.properties

• 测试
  在Kafka集群使用kafka-console-producer.sh生成数据后，就可以在HBase查到数据。

Kafka->OSS

• 创建OSS路径

  创建OSS Bucket及目录，如oss://flume-test/result。

• 配置Flume
  Flume向OSS写入数据时，需要占用较大的JVM内存，可以改小OSS缓存或者增大Flume Agent的Xmx：

  • 修改OSS缓存大小

    将hdfs-site.xml配置文件从/etc/ecm/hadoop-conf拷贝至/etc/ecm/flume-conf，改小配置项 smartdata.cache.buffer.size 的值，例如修改为1048576。

  • 修改Xmx
    在Flume的配置路径/etc/ecm/flume-conf下，复制配置文件flume-env.sh.template并重命名为flume-env.sh，设置Xmx，例如设置为1G。

    export JAVA_OPTS="-Xmx1g"

  创建配置文件flume.properties，添加如下配置。其中，配置项a1.sources.source1.kafka.bootstrap.servers填写Kafka集群broker的host和端口号，a1.sinks.k1.hdfs.path为OSS路径。

  a1.sources = source1
  a1.sinks = k1
  a1.channels = c1
  
  a1.sources.source1.type = org.apache.flume.source.kafka.KafkaSource
  a1.sources.source1.channels = c1
  a1.sources.source1.kafka.bootstrap.servers = kafka-host1:port1,kafka-host2:port2...
  a1.sources.source1.kafka.topics = flume-test
  a1.sources.source1.kafka.consumer.group.id = flume-test-group
  
  a1.sinks.k1.type = hdfs
  a1.sinks.k1.hdfs.path = oss://flume-test/result
  a1.sinks.k1.hdfs.fileType=DataStream
  
  # Use a channel which buffers events in memory
  a1.channels.c1.type = memory
  a1.channels.c1.capacity = 100
  a1.channels.c1.transactionCapacity = 100
  
  # Bind the source and sink to the channel
  a1.sources.source1.channels = c1
  a1.sinks.k1.channel = c1

• 启动Flume
  如果配置Flume时修改了OSS缓存大小，使用classpath参数传入OSS相关依赖和配置。

  flume-ng agent --name a1 --conf /etc/ecm/flume-conf  --conf-file flume.properties  --classpath "/opt/apps/extra-jars/*:/etc/ecm/flume-conf/hdfs-site.xml"

  如果修改了Flume Agent的Xmx，只需传入OSS相关依赖。

  flume-ng agent --name a1 --conf /etc/ecm/flume-conf  --conf-file flume.properties  --classpath "/opt/apps/extra-jars/*"

• 测试

  在Kafka集群使用kafka-console-producer.sh生成数据后，在OSS的oss://flume-test/result路径下会以当前时间的（毫秒）时间戳为后缀生成文件FlumeData.xxxx。

Kerberos Kafka source

跨域互信使用Flume