2025年flume读取kafka写hive

科技前沿 • 2025-01-11 22:37 • 阅读 53

大家好，我是讯享网，很高兴认识大家。

注意：需要提前创建hive接收信息的表--test0617

否则报错:

Caused by: NoSuchObjectException(message:rt1.cr_kafka_t_test0707 table not found)

讯享网

配置文件

localAgent.sources = skafka
localAgent.sinks = shive
localAgent.channels = k2h

#k2h shive
localAgent.sinks.shive.channel = k2h
#skafka k2h
localAgent.sources.skafka.channels = k2h

#describe the source
localAgent.sources.skafka.type = org.apache.flume.source.kafka.KafkaSource
localAgent.sources.skafka.zookeeperConnect = localhost:2181
localAgent.sources.skafka.batchSize = 1000
localAgent.sources.skafka.batchDurationMillis = 500
localAgent.sources.skafka.kafka.bootstrap.servers = localhost:9092
localAgent.sources.skafka.kafka.topics = mytopic0617
localAgent.sources.skafka.kafka.consumer.group.id = kafka007
localAgent.sources.skafka.kafka.consumer.auto.offset.reset = earliest

#use a channel which buffers events in memory
localAgent.channels.k2h.type = memory
localAgent.channels.k2h.capacity = 1000
localAgent.channels.k2h.transactionCapacity = 1000

讯享网

#sinks type hive
localAgent.sinks.shive.type = hive
localAgent.sinks.shive.channel = k2h
localAgent.sinks.shive.hive.metastore = thrift://localhost:9083
localAgent.sinks.shive.hive.database = database1
localAgent.sinks.shive.hive.table = test0617
localAgent.sinks.shive.hive.partition = %Y-%m-%d
localAgent.sinks.shive.batchSize = 500
localAgent.sinks.shive.useLocalTimeStamp = true
localAgent.sinks.shive.round = true
localAgent.sinks.shive.roundValue = 24
localAgent.sinks.shive.roundUnit = hour
localAgent.sinks.shive.serializer = JSON

启动异常：

org.apache.flume.ChannelException: Put queue for MemoryTransaction of capacity 1000 full, consider committing more frequently, increasing capacity or increasing thread count

解决，入上配置中标红部分修改之后，异常解决。

需要启动hdfs、hive、mysql、zk、kafka、flume，注意启动顺序。

2025年flume读取kafka写hive

配置文件

启动异常：

相关推荐