安装环境

Java环境, 略 (Flume依赖Java)Flume下载, 略Scala环境, 略 (Kafka依赖Scala)Kafak下载, 略Hadoop下载, 略 (不需要启动, 写OSS依赖)

配置Hadoop

下载JindoSDK(连接OSS依赖), 下载地址Github 解压后配置环境变量

export JINDOSDK_HOME=/usr/lib/jindosdk-x.x.x

export HADOOP_CLASSPATH=$HADOOP_CLASSPATH:${JINDOSDK_HOME}/lib/*

修改Hadoop配置, core-site.xml

fs.oss.credentials.provider

com.aliyun.jindodata.oss.auth.SimpleCredentialsProvider

fs.oss.accessKeyId

xxxx

fs.oss.accessKeySecret

xxxx

fs.oss.endpoint

xxxxx

fs.AbstractFileSystem.oss.impl

com.aliyun.jindodata.oss.JindoOSS

fs.oss.impl

com.aliyun.jindodata.oss.JindoOssFileSystem

配置可参考非EMR集群接入OSS-HDFS服务快速入门

配置Flume

此部分全文最关键, 请仔细看

基础配置部分, Flume配置

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 = xxx

a1.sources.source1.kafka.topics = test

a1.sources.source1.kafka.consumer.group.id = flume-sink-group # 消费者组, 云组件需要先在管理后台创建

a1.sources.source1.kafka.consumer.auto.offset.reset = earliest # 从头消费Kafka里数据

a1.sinks.k1.channel = c1

a1.sinks.k1.type = hdfs

a1.sinks.k1.hdfs.path = oss://xxx/test/%Y%m%d # 自动按天分文件夹

a1.sinks.k1.hdfs.fileType=DataStream

a1.channels.c1.type = memory

a1.channels.c1.capacity = 1000

a1.channels.c1.transactionCapacity = 1000

可参考使用Flume同步EMR Kafka集群的数据至OSS-HDFS服务 2. 进阶配置, 根据自己情况按需配置

a1.sinks.k1.hdfs.rollInterval = 600 # 5分钟切换一个新文件

a1.sinks.k1.hdfs.rollSize = 134217728 # 或者文件大小达到128M则切换新文件

a1.sinks.k1.hdfs.rollCount = 0 # 写入多少条数据切换新文件, 0为不限制

我这里是为了防止sink的文件过于零碎, 但因为使用的memory channel, 缓存时间过长容易丢数据 3. Flume JVM参数 默认启动时-Xmx20m, 过于小了, 加大堆内存可以直接放开flume-env.sh内JAVA_OPTS的注释

export JAVA_OPTS="-Xms100m -Xmx2000m -Dcom.sun.management.jmxremote"

Channel问题 如果对数据一致性要求较高, 可以把memory channel改用file channel, 请自行研究

XX启动!

几条测试命令

bin/zookeeper-server-start.sh config/zookeeper.properties # 启动zookeeper

bin/kafka-server-start.sh config/server.properties # 启动kafak服务

bin/flume-ng agent --conf conf --conf-file conf/flume-conf.properties --name a1 # 启动flume

bin/kafka-console-producer.sh --topic flume-test --bootstrap-server localhost:9092 # 启动一个生产者写测试数据

文章链接

评论可见,请评论后查看内容,谢谢!!!
 您阅读本篇文章共花了: