Kafka 在行动:7步实现从RDBMS到Hadoop的实时流传输

时间:2021-04-13 19:12:07

对于寻找方法快速吸收数据到Hadoop数据池的企业, Kafka是一个伟大的选择。Kafka是什么? 它是一个分布式,可扩展的可靠消息系统,把采取发布-订阅模型的应用程序/数据流融为一体。 这是Hadoop的技术堆栈中的关键部分,支持实时数据分析或物联网数据货币化。

本文目标读者是技术人员。 继续读,我会图解Kafka如何从关系数据库管理系统(RDBMS)里流输数据到Hive, 这可以提供一个实时分析使用案例。 为了参考方便,本文使用的组件版本是Hive 1.2.1,Flume 1.6和Kafka 0.9。

第 1 段(可获 1.5 积分)

如果你想看一下Kafka是什么和其用途的概述, 看看我 在Datafloq 上发布的一篇早期博客。

Kafka用武之地:整体解决方案架构

下图显示了在整体解决方案架构中,RDBMS的业务数据传递到目标 Hive 表格结合了 Kafka , Flume和Hive交易功能。

Kafka 在行动:7步实现从RDBMS到Hadoop的实时流传输

7步实时流传输到Hadoop

现在深入到解决方案的详细信息,我会告诉你如何简单几步实时流输数据到Hadoop。

1. 从关系数据库管理系统(RDBMS)提取数据

第 2 段(可获 1.34 积分)

所有关系数据库都有一个记录最近交易的日志文件。 我们的传输流解决方案的第一步是,在能够传到Hadoop的信息格式中获得这些交易。 讲完提取机制得单独占用一篇博文--所以 如果你想了解更多此过程的信息, 请联系我们。

2. 建立Kafka Producer

发布消息到Kafka主题的过程被称为“生产者”。“主题”是Kafka保存的分类消息。 RDBMS的交易将被转换为Kafka话题。 对于该例,让我们想一想销售团队的数据库,其中的交易是作为Kafka主题发表的。 建立Kafka生产者需要以下步骤:

第 3 段(可获 1.65 积分)
$ cd /usr/hdp/2.4.0.0-169/kafka
$ bin/kafka-topics.sh --create --zookeeper sandbox.hortonworks.com:2181 --replication-factor 1 --partitions 1 --topic SalesDBTransactions Created topic "SalesDBTransactions".
$ bin/kafka-topics.sh --list --zookeeper sandbox.hortonworks.com:2181
SalesDBTransactions

3. 设置 Hive

接下来,我们将在Hive中创建一张表,准备接收销售团队的数据库事务。 在这个例子中,我们将创建一个客户表:

[bedrock@sandbox ~]$ beeline -u jdbc:hive2:// -n hive -p hive
0: jdbc:hive2://> use raj;
create table customers (id string, name string, email string, street_address string, company string) partitioned by (time string)
clustered by (id) into 5 buckets stored as orc
location '/user/bedrock/salescust'
TBLPROPERTIES ('transactional'='true');
第 4 段(可获 0.39 积分)

为了让Hive能够处理交易, 配置中需要以下设置:

hive.txn.manager = org.apache.hadoop.hive.ql.lockmgr.dbtxnmanager

4.设置Flume Agent,从Kafka到Hive流传输

现在让我们来看看如何创建Flume代理,实现从Kafka主题中获取数据,发送到Hive表。

遵循步骤来设置环境,然后建立Flume代理:

$ pwd
/home/bedrock/streamingdemo
$ mkdir flume/checkpoint
$ mkdir flume/data
$ chmod 777 -R flume $ export HIVE_HOME=/usr/hdp/current/hive-server2
$ export HCAT_HOME=/usr/hdp/current/hive-webhcat
 
$ pwd
/home/bedrock/streamingdemo/flume
$ mkdir logs
第 5 段(可获 0.79 积分)

接着,如下创建一个log4j属性文件:

[bedrock@sandbox conf]$ vi log4j.properties

flume.root.logger=INFO,LOGFILE
flume.log.dir=/home/bedrock/streamingdemo/flume/logs
flume.log.file=flume.log

然后为Flume代理使用下面的配置文件:

$ vi flumetohive.conf
flumeagent1.sources = source_from_kafka
flumeagent1.channels = mem_channel
flumeagent1.sinks = hive_sink # Define / Configure source
flumeagent1.sources.source_from_kafka.type = org.apache.flume.source.kafka.KafkaSource flumeagent1.sources.source_from_kafka.zookeeperConnect = sandbox.hortonworks.com:2181 flumeagent1.sources.source_from_kafka.topic = SalesDBTransactions
flumeagent1.sources.source_from_kafka.groupID = flume
flumeagent1.sources.source_from_kafka.channels = mem_channel
flumeagent1.sources.source_from_kafka.interceptors = i1
flumeagent1.sources.source_from_kafka.interceptors.i1.type = timestamp
flumeagent1.sources.source_from_kafka.consumer.timeout.ms = 1000
 
# Hive Sink
flumeagent1.sinks.hive_sink.type = hive
flumeagent1.sinks.hive_sink.hive.metastore = thrift://sandbox.hortonworks.com:9083 flumeagent1.sinks.hive_sink.hive.database = raj
flumeagent1.sinks.hive_sink.hive.table = customers
flumeagent1.sinks.hive_sink.hive.txnsPerBatchAsk = 2
flumeagent1.sinks.hive_sink.hive.partition = %y-%m-%d-%H-%M
flumeagent1.sinks.hive_sink.batchSize = 10
flumeagent1.sinks.hive_sink.serializer = DELIMITED
flumeagent1.sinks.hive_sink.serializer.delimiter = ,
flumeagent1.sinks.hive_sink.serializer.fieldnames = id,name,email,street_address,company # Use a channel which buffers events in memory
flumeagent1.channels.mem_channel.type = memory
flumeagent1.channels.mem_channel.capacity = 10000
flumeagent1.channels.mem_channel.transactionCapacity = 100 # Bind the source and sink to the channel
flumeagent1.sources.source_from_kafka.channels = mem_channel
flumeagent1.sinks.hive_sink.channel = mem_channel
第 6 段(可获 0.23 积分)

5.开启Flume代理

使用如下命令开启Flume代理:

$ /usr/hdp/apache-flume-1.6.0/bin/flume-ng agent -n flumeagent1 -f ~/streamingdemo/flume/conf/flumetohive.conf Kafka 在行动:7步实现从RDBMS到Hadoop的实时流传输

6.开启Kafka Stream

如下示例,是一个模拟交易消息, 在实际系统中需要由源数据库生成。 例如,以下可能来自重复SQL交易的Oracle数据流,这些交易已提交到数据库, 也可能来自GoledenGate。

$ cd /usr/hdp/2.4.0.0-169/kafka
$ bin/kafka-console-producer.sh --broker-list sandbox.hortonworks.com:6667 --topic SalesDBTransactions 1,"Nero Morris","porttitor.interdum@Sedcongue.edu","P.O. Box 871, 5313 Quis Ave","Sodales Company" 2,"Cody Bond","ante.lectus.convallis@antebibendumullamcorper.ca","232-513 Molestie Road","Aenean Eget Magna Incorporated" 3,"Holmes Cannon","a@metusAliquam.edu","P.O. Box 726, 7682 Bibendum Rd.","Velit Cras LLP" 4,"Alexander Lewis","risus@urna.edu","Ap #375-9675 Lacus Av.","Ut Aliquam Iaculis Inc." 5,"Gavin Ortiz","sit.amet@aliquameu.net","Ap #453-1440 Urna. St.","Libero Nec Ltd" 6,"Ralph Fleming","sociis.natoque.penatibus@quismassaMauris.edu","363-6976 Lacus. St.","Quisque Fringilla PC" 7,"Merrill Norton","at.sem@elementum.net","P.O. Box 452, 6951 Egestas. St.","Nec Metus Institute" 8,"Nathaniel Carrillo","eget@massa.co.uk","Ap #438-604 Tellus St.","Blandit Viverra Corporation" 9,"Warren Valenzuela","tempus.scelerisque.lorem@ornare.co.uk","Ap #590-320 Nulla Av.","Ligula Aliquam Erat Incorporated" 10,"Donovan Hill","facilisi@augue.org","979-6729 Donec Road","Turpis In Condimentum Associates" 11,"Kamal Matthews","augue.ut@necleoMorbi.org","Ap #530-8214 Convallis, St.","Tristique Senectus Et Foundation"
第 7 段(可获 0.88 积分)

Kafka 在行动:7步实现从RDBMS到Hadoop的实时流传输

7.接收Hive数据

以上所有完成, 现在从Kafka发送数据, 你会看到,几秒之内,数据流就发送到Hive表了。