通过Flume简单实现Kafka与Hive对接（Json格式）

{"id":1,"name":"小李"}
{"id":2,"name":"小张"}
{"id":3,"name":"小刘"}
{"id":4,"name":"小王"}

    <property>
   	 <name>hive.txn.manager</name>
   	 <value>org.apache.hadoop.hive.ql.lockmgr.DbTxnManager</value>
    </property>
    <property>
    	<name>hive.support.concurrency</name>
     	<value>true</value>
    </property>
    <property>
    	<name>hive.metastore.uris</name>
    	<value>thrift://localhost:9083</value>
    </property>

hive> create database hivetokafka;
 
hive> create table kafkatable(id int,name string) 
hive> clustered by(id) into 2 buckets stored as orc tblproperties('transactional'='true');

 hive --service metastore & 

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


#kafka为souce的配置
a.sources.source_from_kafka.type=org.apache.flume.source.kafka.KafkaSource
a.sources.source_from_kafka.zookeeperConnect=localhost:2181
a.sources.source_from_kafka.bootstrap.servers=localhost:9092
a.sources.source_from_kafka.topic=testtopic
a.sources.source_from_kafka.channels=mem_channel
a.sources.source_from_kafka.consumer.timeout.ms=1000
#hive为sink的配置
a.sinks.hive_sink.type=hive
a.sinks.hive_sink.hive.metastore=thrift://localhost:9083
a.sinks.hive_sink.hive.database=hivetokafka
a.sinks.hive_sink.hive.table=kafkatable
a.sinks.hive_sink.hive.txnsPerBatchAsk=2
a.sinks.hive_sink.batchSize=10
a.sinks.hive_sink.serializer=JSON
a.sinks.hive_sink.serializer.fieldnames=id,name
#channel的配置
a.channels.mem_channel.type=memory
a.channels.mem_channel.capacity=1500
a.channels.mem_channel.transactionCapacity=1000
#三者之间的关系
a.sources.source_from_kafka.channels=mem_channel
a.sinks.hive_sink.channel=mem_channel

flume-ng agent --conf conf/ --conf-file conf/….  --name a -Dflume.root.logger=INFO,console;

bin/flume-ng agent --conf myconf/ --conf-file myconf/kafkatohive.conf  --name a -Dflume.root.logger=INFO,console;

kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic testtopic

kafka-console-producer.sh --broker-list localhost:9092 --topic testtopic

>{"id":1,"name":"小李"}
>{"id":2,"name":"小张"}
>{"id":3,"name":"小刘"}
>{"id":4,"name":"小王"}

hive> select * from student;
OK
1	小李
2	小张
3	小刘
4	小王

Time taken: 0.589 seconds, Fetched: 10 row(s)