E-MapReduce（简称EMR） 从EMR-3.16.0版本开始支持Apache Flume。本文介绍如何使用Flume将数据从EMR Kafka集群同步至EMR
Hadoop集群的HDFS、Hive、HBase以及阿里云OSS。
前提条件
- 已开通OSS服务，详情请参见开通OSS服务。
- 已创建Hadoop集群，并且选择了Flume和HBase服务，详情请参见创建集群。
说明 Flume软件安装目录在
/usr/lib/flume-current下，其他常用文件路径获取方式请参见
常用文件路径。
- 已创建Kafka集群，详情请参见创建集群。
同步Kafka数据至HDFS
- 通过SSH方式连接Hadoop集群。
- 配置Flume。
- 进入目录/etc/ecm/flume-conf。
- 创建配置文件flume.properties。
- 添加如下内容。
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.sources.source1.kafka.bootstrap.servers：Kafka集群Broker的Host和端口号。
- a1.sources.source1.kafka.topics：Flume消费Kafka数据的Topic。
- a1.channels.c1.capacity：通道中存储的最大事件数。请根据实际环境修改该参数值。
- a1.channels.c1.transactionCapacity：每个事务通道将从源接收或提供给接收器的最大事件数。请根据实际环境修改该参数值。
- a1.sinks.k1.hdfs.path：Flume向HDFS写入数据的路径。
如果配置项
a1.sinks.k1.hdfs.path使用URL的形式，针对不同集群代码示例如下：
- 执行如下命令启动服务。
flume-ng agent --name a1 --conf /etc/ecm/flume-conf --conf-file flume.properties
- 测试数据情况。
- 通过SSH方式连接Kafka集群，详情请参见使用SSH连接主节点。
- 创建名称为flume-test的Topic。
/usr/lib/kafka-current/bin/kafka-topics.sh --partitions 10 --replication-factor 2 --zookeeper emr-header-1:2181 /kafka-1.0.0 --topic flume-test --create
- 生成测试数据。
kafka-console-producer.sh --topic flume-test --broker-list emr-header-1:9092
例如输入abc并回车。
- 通过SSH方式连接Hadoop集群，查看生成的文件。
Flume会在HDFS中以当前时间的（毫秒）时间戳生成文件
FlumeData.xxxx。文件内容是在Kafka中输入的数据
abc。
hdfs dfs -cat /tmp/flume/test-data/<FlumeData.xxxx>
同步Kafka数据至Hive
- 连接Hadoop集群并创建Hive表。
Flume使用事务操作将数据写入Hive，需要在创建Hive表（flume_test）时设置transactional属性。
create table flume_test2 (id int, content string)
clustered by (id) into 2 buckets stored as orc TBLPROPERTIES('transactional'='true');
- 配置Flume。
- 进入目录/etc/ecm/flume-conf。
- 创建配置文件flume.properties。
- 添加如下内容。
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
- a1.sources.source1.kafka.bootstrap.servers：Kafka集群Broker的Host和端口号。
- a1.channels.c1.capacity：通道中存储的最大事件数。请根据实际环境修改该参数值。
- a1.channels.c1.transactionCapacity：每个事务通道将从源接收或提供给接收器的最大事件数。请根据实际环境修改该参数值。
- a1.sinks.k1.hive.metastore：Hive metastore的URI，格式为thrift://emr-header-1.cluster-xxx:9083。其中emr-header-1.cluster-xxx您可以通过
hostname获取。
- 执行如下命令启动服务。
flume-ng agent --name a1 --conf /etc/ecm/flume-conf --conf-file flume.properties
- 测试数据情况。
- 通过SSH方式连接Kafka集群，详情请参见使用SSH连接主节点。
- 创建名称为flume-test的Topic。
/usr/lib/kafka-current/bin/kafka-topics.sh --partitions 10 --replication-factor 2 --zookeeper emr-header-1:2181 /kafka-1.0.0 --topic flume-test --create
- 生成测试数据。
kafka-console-producer.sh --topic flume-test --broker-list emr-header-1:9092
例如输入1,a并回车。
- 通过SSH方式连接Hadoop集群，在客户端进行配置并查询Hive事务表。
set hive.support.concurrency=true;
set hive.exec.dynamic.partition.mode=nonstrict;
set hive.txn.manager=org.apache.hadoop.hive.ql.lockmgr.DbTxnManager;
配置好后查询
flume_test表中的数据。
select * from flume_test;
同步Kafka数据至HBase
- 创建HBase表flume_test及列簇column。
create 'flume_test','column'
- 配置Flume。
- 进入目录/etc/ecm/flume-conf。
- 创建配置文件flume.properties。
- 添加如下内容。
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 = <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.sources.source1.kafka.bootstrap.servers：Kafka集群Broker的Host和端口号。
- a1.sinks.k1.table：HBase表名。
- a1.sinks.k1.columnFamily：列簇名。
- a1.channels.c1.capacity：通道中存储的最大事件数。请根据实际环境修改该参数值。
- a1.channels.c1.transactionCapacity：每个事务通道将从源接收或提供给接收器的最大事件数。请根据实际环境修改该参数值。
- 启动服务。
flume-ng agent --name a1 --conf /etc/ecm/flume-conf --conf-file flume.properties
- 测试数据写入情况。
在Kafka集群使用
kafka-console-producer.sh生成数据后，在HBase查到数据。
同步Kafka数据至OSS
- 创建OSS路径。
创建存储空间，详情请参见
创建存储空间。
本文路径为oss://flume-test/result。
- 配置Flume。
- 修改OSS缓存大小或设置JVM最大可用内存（Xmx）。
Flume向OSS写入数据时，因为需要占用较大的JVM内存，所以可以改小OSS缓存或者增大Flume Agent的Xmx。
- 创建配置文件flume.properties。
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
- a1.sources.source1.kafka.bootstrap.servers：Kafka集群Broker的Host和端口号。
- a1.sinks.k1.hdfs.path：OSS路径。
- a1.channels.c1.capacity：通道中存储的最大事件数。请根据实际环境修改该参数值。
- a1.channels.c1.transactionCapacity：每个事务通道将从源接收或提供给接收器的最大事件数。请根据实际环境修改该参数值。
- 启动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
消费高安全Kafka集群的数据时，需要完成额外的配置：
- 在Kafka集群配置Kerberos认证，将生成的keytab文件test.keytab拷贝至Hadoop集群的/etc/ecm/flume-conf路径下，详情请参见兼容MIT Kerberos认证；将Kafka集群的/etc/ecm/has-conf/krb5.conf文件拷贝至Hadoop集群的/etc/ecm/flume-conf路径下。
- 配置flume.properties。
在
flume.properties中添加如下配置。
a1.sources.source1.kafka.consumer.security.protocol = SASL_PLAINTEXT
a1.sources.source1.kafka.consumer.sasl.mechanism = GSSAPI
a1.sources.source1.kafka.consumer.sasl.kerberos.service.name = kafka
- 配置Kafka client。
- 在/etc/ecm/flume-conf下创建文件flume_jaas.conf。
KafkaClient {
com.sun.security.auth.module.Krb5LoginModule required
useKeyTab=true
storeKey=true
keyTab="/etc/ecm/flume-conf/test.keytab"
serviceName="kafka"
principal="test@EMR.${realm}.COM";
};
${realm} 需要替换为Kafka集群的Kerberos realm。
${realm}获取方式：在Kafka集群执行命令hostname，得到形式为emr-header-1.cluster-xxx的主机名，例如emr-header-1.cluster-123456，其中数字串123456即为realm。
- 修改/etc/ecm/flume-conf/flume-env.sh。
初始情况下，
/etc/ecm/flume-conf/下没有
flume-env.sh 文件，需要拷贝
flume-env.sh.template并重命名为
flume-env.sh。添加如下内容。
export JAVA_OPTS="$JAVA_OPTS -Djava.security.krb5.conf=/etc/ecm/flume-conf/krb5.conf"
export JAVA_OPTS="$JAVA_OPTS -Djava.security.auth.login.config=/etc/ecm/flume-conf/flume_jaas.conf"
- 设置域名。
将Kafka集群各节点的长域名和IP的绑定信息添加到Hadoop集群的
/etc/hosts。长域名的形式例如
emr-header-1.cluster-xxxx。
说明 图中标注①表示的是Hadoop集群的域名；图中标注②表示新增加的Kafka集群域名。
跨域互信使用Flume
在配置了跨域互信后，其他配置如下：
- 在Kafka集群配置Kerberos认证，将生成的keytab文件test.keytab拷贝至Hadoop集群的/etc/ecm/flume-conf路径下，详情请参见兼容MIT Kerberos认证。
- 配置flume.properties。
在
flume.properties中添加如下配置。
a1.sources.source1.kafka.consumer.security.protocol = SASL_PLAINTEXT
a1.sources.source1.kafka.consumer.sasl.mechanism = GSSAPI
a1.sources.source1.kafka.consumer.sasl.kerberos.service.name = kafka
- 配置Kafka client。
- 在/etc/ecm/flume-conf下创建文件flume_jaas.conf，内容如下。
KafkaClient {
com.sun.security.auth.module.Krb5LoginModule required
useKeyTab=true
storeKey=true
keyTab="/etc/ecm/flume-conf/test.keytab"
serviceName="kafka"
principal="test@EMR.${realm}.COM";
};
${realm}替换为Kafka集群的Kerberos realm。
${realm}获取方式：在Kafka集群执行命令hostname，得到形式为emr-header-1.cluster-xxx的主机名，例如emr-header-1.cluster-123456，其中数字串123456即为realm。
- 修改/etc/ecm/flume-conf/flume-env.sh。
初始情况下，
/etc/ecm/flume-conf/下没有
flume-env.sh文件，需要拷贝
flume-env.sh.template并重命名为
flume-env.sh。添加如下内容。
export JAVA_OPTS="$JAVA_OPTS -Djava.security.auth.login.config=/etc/ecm/flume-conf/flume_jaas.conf""
在文档使用中是否遇到以下问题
更多建议
匿名提交