Help & Documentation>Elastic MapReduce>EMR Development Guide>Flume Development Guide>Storing Kafka Data in Hive Through Flume

Storing Kafka Data in Hive Through Flume

Last updated: 2023-12-26 14:41:55

On this page:

Description

Collect data from Kafka via Flume and store it in Hive.

Development Preparations

As the task requires access to Tencent Cloud's message queue CKafka, it is necessary to first create a CKafka instance, as detailed in Message Queue CKafka.
Ensure that you have activated Tencent Cloud and have created an EMR cluster. When creating the EMR cluster, you need to select the Flume component in the software configuration interface.

Utilizing the Kafka toolkit in the EMR cluster

Initially, it is necessary to view the internal network IP and port number of CKafka. Log into the Message Queue CKafka console, select the CKafka instance you wish to use, and view its internal network IP as $kafkaIP in Basic Information. The port number is generally set to 9092 by default. Create a new topic named kafka_test in the topic management interface.

Configuring Flume

1. Create the Flume configuration file hive_kafka.properties
vim hive_kafka.properties
agent.sources = kafka_source
agent.channels = mem_channel
agent.sinks = hive_sink
# Configuring the following source
agent.sources.kafka_source.type = org.apache.flume.source.kafka.KafkaSource
agent.sources.kafka_source.channels = mem_channel
agent.sources.kafka_source.batchSize = 5000
agent.sources.kafka_source.kafka.bootstrap.servers = $kafkaIP:9092
agent.sources.kafka_source.kafka.topics = kafka_test
# Configuring the following sink
agent.sinks.hive_sink.channel = mem_channel
agent.sinks.hive_sink.type = hive
agent.sinks.hive_sink.hive.metastore = thrift://172.16.32.51:7004
agent.sinks.hive_sink.hive.database = default
agent.sinks.hive_sink.hive.table = weblogs
agent.sinks.hive_sink.hive.partition = asia,india,%y-%m-%d-%H-%M
agent.sinks.hive_sink.useLocalTimeStamp = true
agent.sinks.hive_sink.round = true
agent.sinks.hive_sink.roundValue = 10
agent.sinks.hive_sink.roundUnit = minute
agent.sinks.hive_sink.serializer = DELIMITED
agent.sinks.hive_sink.serializer.delimiter = ","
agent.sinks.hive_sink.serializer.serdeSeparator = ','
agent.sinks.hive_sink.serializer.fieldnames =id,msg
# Configuring the following channel
agent.channels.mem_channel.type = memory
agent.channels.mem_channel.capacity = 100000
agent.channels.mem_channel.transactionCapacity = 100000
The hive.metastore can be confirmed in the following manner:
grep "hive.metastore.uris" -C 2 /usr/local/service/hive/conf/hive-site.xml
<property>
<name>hive.metastore.uris</name>
<value>thrift://172.16.32.51:7004</value>
</property>
2. Create a Hive table
create table weblogs ( id int , msg string )
partitioned by (continent string, country string, time string)
clustered by (id) into 5 buckets
stored as orc TBLPROPERTIES ('transactional'='true');
Note
It is imperative that the table is partitioned and bucketed, stored as ORC, and set with TBLPROPERTIES ('transactional'='true'). None of these conditions can be omitted.
3. Initiate Hive transaction Add the following configuration items to hive-site.xml via the console.
<property>
<name>hive.support.concurrency</name>
<value>true</value>
</property>
<property>
<name>hive.exec.dynamic.partition.mode</name>
<value>nonstrict</value>
</property>
<property>
<name>hive.txn.manager</name>
<value>org.apache.hadoop.hive.ql.lockmgr.DbTxnManager</value>
</property>
<property>
<name>hive.compactor.initiator.on</name>
<value>true</value>
</property>
<property>
<name>hive.compactor.worker.threads</name>
<value>1</value>
</property>
<property>
<name>hive.enforce.bucketing</name>
<value>true</value>
</property>
Note
After the configuration is issued and restarted, the hadoop-hive log will indicate that the metastore cannot connect. Please disregard this error. Due to the process startup sequence, the metastore needs to be launched before hiveserver2.
4. Copy Hive's hive-hcatalog-streaming-xxx.jar to the lib directory of Flume.
cp -ra /usr/local/service/hive/hcatalog/share/hcatalog/hive-hcatalog-streaming-2.3.3.jar /usr/local/service/flume/lib/
5. Execute Flume
./bin/flume-ng agent --conf ./conf/ -f hive_kafka.properties -n agent -Dflume.root.logger=INFO,console
6. Execute Kafka producer
[hadoop@172 kafka]$ ./bin/kafka-console-producer.sh --broker-list $kafkaIP:9092 --topic kafka_test
1,hello
2,hi

Test

Enter information in the Kafka producer client and press return.
Observe whether there is corresponding data in the Hive table.

Reference

Configuration Guide for Hive-Sink
Configuration Guide for Hive Logs
中国站 - English