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

将以下存储在kafka的topic中的JSON格式字符串，对接存储到Hive的表中

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

1、在hive/conf/hive-site.xml中添加或修改如下内容：

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

2、创建database、table，其中表有id、name这个两个字段

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');

3、执行 hive --service metastore & 启动元数据服务

 hive --service metastore &

4、配置conf文件，这里文件名和位置可以随意（我的是放在hive/myconf/新建的目录下，名字为kafkatohive.conf），添加如下内容

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

5、将/hive/hcatalog/share/hcatalog/hive-hcatalog-streaming-x.x.x.jar拷贝到/flume/lib/下

此外还需要注意/hive/lib/guava-xx.x-jre.jar下与/flume/lib/下的版本是否一致。

6、启动flume，命令格式如下

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

我这里就是(在flume/路径下 )：

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

7、新建终端窗口，创建topic（默认已经启动了zookeeper和kafka服务了）

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

8、启动kafka生产者，进行生产消息

启动命令：

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

生产消息：

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

9、查看结果

hive> select * from student;
OK
1   小李
2   小张
3   小刘
4   小王Time taken: 0.589 seconds, Fetched: 10 row(s)

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

hive 导出json格式文件_hive中创建hive-json格式的表及查询
在hive中对于json的数据格式,可以使用get_json_object或json_tuple先解析然后查询. 也可以直接在hive中创建json格式的表结构,这样就可以直接查询,实战如下(hive ...
hive 导出json格式文件_hive支持json格式的数据
Hive 支持完全 json 格式的数据  现有 json 格式的数据 test.txt ,如下 {"name":"zhang","age" ...
hive中json格式字段解析及map使用
json 如果hive表中有扩展字段,那么扩展字段格式最常见的就是json格式,所以如何解析json字符串相对繁琐(虽然没啥技术
Hive解析Json格式用户日志
解析json格式数据 Json数据 HIVE SQL解析第一种方式第二种方式第三种方式 string类型的数组形式的行转列拆分 Json数据第一次写博客,mark一下 --20191025 下 ...
hive 导出json格式文件_hive存储json格式文件
hive从0.12版本以后就开始自身支持json文件的格式了 1.文件格式下面是测试用的文件预览格式,我将其存储为jsonTest.json文件,放在了/root目录下 {"_locati ...
hive 导出json格式文件_Hive 系列之基本操作合集
下面是本课程概览: (1)hive系列之简介,安装,beeline和hiveserver2 (2)hive系列之基本操作 (3)hive系列之udf (4)hive系列之二级分区和动态分区 (5)hi ...
hive 导出json格式文件_Magicodes.IE在.NET Core中通过请求头导出多种格式文件
前言在2.2里程碑中我们增加了一些新的功能,正如标题所写通过请求头进行导出我们不同格式的文件.下面我们来看一下如何使用.通过这种方式无论是对我们的数据多用途,还是说对我们的数据校验都做到了轻松易配. ...
Storm集成HBase、JDBC、Kafka、Hive测试
/*** storm集成Kafka.Hive.JDBC.HBase.HDFS* Created by sker on 17-11-13* kafka集成storm,将数据发到JobBolt做中文分词逻 ...
Storm集成HBase、JDBC、Kafka、Hive
代码参上 /*** storm集成Kafka.Hive.JDBC.HBase.HDFS* Created by sker on 17-11-13* kafka集成storm,将数据发到JobBolt做 ...

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

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

最新文章

热门文章