利用Flume、kafka与hive对接

淡淡的烟草味﹌ 2022-01-11 09:49 385阅读 0赞

Flume是Cloudera提供的一个高可用的，高可靠的，分布式的海量日志采集，聚合和传输的系统，Flume支持系统中定制的各类数据发送方，用于收集数据，同时，Flume提供对数据进行简单处理，并写到各种数据接受方（可定制的）的能力。

当前Flume有两个版本Flume 0.9X版本统称Flume-og，Flume1.X版本的统称Flume-ng。由于Flume-ng经过重大重构，与Flume-og有很大不同。使用时请注意区分。

接下来就是操作步骤了。

# 1.hive建表 #

注意:建表时要进行分桶，赋予事务性，这样才可以对hive进行增删改查。默认是不被允许的。

建表如下：

create table kafkatest(id int,name string,age int) clustered by(id) into 2 buckets 
    stored as orc tblproperties(‘transactional’=’true’);

# 2.配置flume（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=kafkatest
    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=itcast
    a.sinks.hive_sink.hive.table=kafkatest
    a.sinks.hive_sink.hive.txnsPerBatchAsk=2
    a.sinks.hive_sink.batchSize=10
    a.sinks.hive_sink.serializer=DELIMITED
    a.sinks.hive_sink.serializer.delimiter=,
    a.sinks.hive_sink.serializer.fieldnames=id,name,age
    #channel的配置
    a.channels.mem_channel.type=memory
    a.channels.mem_channel.capacity=1000
    a.channels.mem_channel.transactionCapacity=100
    #三者之间的关系
    a.sources.source_from_kafka.channels=mem_channel
    a.sinks.hive_sink.channel=mem_channel

# 3.启动hive前要设置hive.txn.manager #

第一种方式：脚本命令

set hive.txn.manager=org.apache.hadoop.hive.ql.lockmgr.DbTxnManager;

第二种方式：配置文件（hive-site.xml）

<property>
        <name>hive.txn.manager</name>
        <value>org.apache.hadoop.hive.ql.lockmgr.DbTxnManager</value>
    </property>

# 4.启动hive #

先执行如下脚本：

hive --service metastore &

再启动hive

# 5.启动flume #

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

# 6.创建kafka的topic #