CDH-Flume从Kafka同步数据到hive

2021-12-16 15:30:01 阅读：228 来源： 互联网

标签：Flume sinks CDH hive kafka source sink channel

启动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

本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享；
2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关；
3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关；
4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除；
5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。

ICode9

CDH-Flume从Kafka同步数据到hive