原文：
对于寻找方法快速吸收数据到Hadoop数据池的企业， Kafka是一个伟大的选择。Kafka是什么？ 它是一个分布式，可扩展的可靠消息系统，把采取发布-订阅模型的应用程序/数据流融为一体。 这是Hadoop的技术堆栈中的关键部分，支持实时数据分析或物联网数据货币化。
本文目标读者是技术人员。 继续读，我会图解Kafka如何从关系数据库管理系统（RDBMS）里流输数据到Hive， 这可以提供一个实时分析使用案例。 为了参考方便，本文使用的组件版本是Hive 1.2.1，Flume 1.6和Kafka 0.9。
如果你想看一下Kafka是什么和其用途的概述， 看看我 上发布的一篇早期博客。
Kafka用武之地：整体解决方案架构
下图显示了在整体解决方案架构中，RDBMS的业务数据传递到目标 表格结合了 ， 和Hive交易功能。
7步实时流传输到Hadoop
现在深入到解决方案的详细信息，我会告诉你如何简单几步实时流输数据到Hadoop。
1. 从关系数据库管理系统（RDBMS）提取数据
所有关系数据库都有一个记录最近交易的日志文件。 我们的传输流解决方案的第一步是，在能够传到Hadoop的信息格式中获得这些交易。 讲完提取机制得单独占用一篇博文--所以 请联系我们。
2. 建立Kafka Producer
发布消息到Kafka主题的过程被称为“生产者”。“主题”是Kafka保存的分类消息。 RDBMS的交易将被转换为Kafka话题。 对于该例，让我们想一想销售团队的数据库，其中的交易是作为Kafka主题发表的。 建立Kafka生产者需要以下步骤：
$ 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 hive0: 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'); 
为了让Hive能够处理交易， 配置中需要以下设置：
hive.txn.manager = org.apache.hadoop.hive.ql.lockmgr.dbtxnmanager
4.设置Flume Agent，从Kafka到Hive流传输
现在让我们来看看如何创建Flume代理，实现从Kafka主题中获取数据，发送到Hive表。
遵循步骤来设置环境，然后建立Flume代理：
接着，如下创建一个log4j属性文件：
[bedrock@sandbox conf]$ vi log4j.propertiesflume.root.logger=INFO,LOGFILEflume.log.dir=/home/bedrock/streamingdemo/flume/logsflume.log.file=flume.log
然后为Flume代理使用下面的配置文件：
$ vi flumetohive.confflumeagent1.sources = source_from_kafkaflumeagent1.channels = mem_channelflumeagent1.sinks = hive_sink# Define / Configure sourceflumeagent1.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
5.开启Flume代理
使用如下命令开启Flume代理：
$ /usr/hdp/apache-flume-1.6.0/bin/flume-ng agent -n flumeagent1 -f ~/streamingdemo/flume/conf/flumetohive.conf 
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.接收Hive数据
以上所有完成， 现在从Kafka发送数据， 你会看到，几秒之内，数据流就发送到Hive表了。