CDH-Flume从Kafka同步数据到hive

启动Flume命令

flume-ng agent -n a -c /opt/cloudera/parcels/CDH-6.3.0-1.cdh6.3.0.p0.1279813/lib/flume-ng/conf/ -f ./kafka2hiveTest.conf -Dflume.root.logger=INFO,console

hive建表 语句

#分桶开启事务并分区
create table log_test(ip string,username string,request_url string) partitioned by (send_date string) clustered by(ip) into 2 buckets stored as orc tblproperties('transactional'='true');

配置文件

a.sources=source_from_kafka
a.channels=mem_channel
a.sinks=hive_sink
a.sinks=k1

#kafka为souce的配置
a.sources.source_from_kafka.type=org.apache.flume.source.kafka.KafkaSource
a.sources.source_from_kafka.kafka.bootstrap.servers=hadoop001:9092,hadoop002:9092,hadoop003:9092,hadoop004:9092
a.sources.source_from_kafka.topic=test-toHive
a.sources.source_from_kafka.batchSize=10
a.sources.source_from_kafka.setTopicHeader=false

#hive为sink的配置
a.sinks.hive_sink.type=hive
a.sinks.hive_sink.hive.metastore=thrift://hadoop001:9083
a.sinks.hive_sink.hive.database=test
a.sinks.hive_sink.hive.table=log_test
a.sinks.hive_sink.hive.partition = %{key},
a.sinks.hive_sink.serializer=JSON

a.sinks.hive_sink.batchSize=10
a.sinks.hive_sink.hive.txnsPerBatchAsk=2
a.sinks.hive_sink.serializer.fieldnames=ip,username,request_url


a.sinks.k1.type=logger



#channel的配置
a.channels.mem_channel.type=memory

#三者之间的关系
a.sources.source_from_kafka.channels=mem_channel
#a.sinks.hive_sink.channel=mem_channel
a.sinks.k1.channel = mem_channel

报错1

Exception in thread "PollableSourceRunner-KafkaSource-source_from_kafka" java.lang.OutOfMemoryError: GC overhead limit exceeded

Exception in thread "SinkRunner-PollingRunner-DefaultSinkProcessor" java.lang.OutOfMemoryError: GC overhead limit exceeded

这是内存溢出问题，主要就是因为jvm的内存设小了。

这个参数需要在“flume-env.sh”文件中进行设置，具体位置可以搜索一下，如果没有这个文件的话，可以通过“flume-env.sh.template”进行复制 ，然后编辑，加上对应内存设置就可以了

export JAVA_OPTS="-Xms2048m -Xmx2048m -Xss256k -Xmn1g -XX:+UseParNewGC -XX:+UseConcMarkSweepGC -XX:-UseGCOverheadLimit"

报错2

java.lang.ClassNotFoundException: org.apache.hadoop.hive.cli.CliSessionState

有报一系列的找不到某个类的错误，解决办法是找到对应的jar包，然后放到lib目录下

/opt/cloudera/parcels/CDH-6.3.0-1.cdh6.3.0.p0.1279813/lib/flume-ng/lib/

/data/opt/cloudera/parcels/CDH-6.3.0-1.cdh6.3.0.p0.1279813/lib/flume-ng/lib/

opt是本地启动时依赖包的位置，data是集群启动时放的位置

标签：Flume,sinks,CDH,hive,kafka,source,sink,channel
来源： https://blog.csdn.net/lp840312696/article/details/121966633