1. 概述
本手册主要介绍了，一个将传统数据接入到Hadoop集群的数据接入方案和实施方法。供数据接入和集群运维人员参考。
1.1. 整体方案
Flume作为日志收集工具，监控一个文件目录或者一个文件，当有新数据加入时，收集新数据发送给Kafka。Kafka用来做数据缓存和消息订阅。Kafka里面的消息可以定时落地到HDFS上，也可以用Spark Streaming来做实时处理，然后将处理后的数据落地到HDFS上。
1.2. 数据接入流程
本数据接入方案，分为以下几个步骤：
l 安装部署Flume：在每个数据采集节点上安装数据采集工具Flume。详见“2、安装部署Flume”。
l 数据预处理：生成特定格式的数据，供Flume采集。详见“3、数据预处理”。
l Flume采集数据到Kafka: Flume采集数据并发送到Kafka消息队列。详见“4、Flume采集数据到Kafka”。
l Kafka数据落地：将Kafka数据落地到HDFS。详见“5、Kafka数据落地”。
2. 安装部署Flume
若要采集数据节点的本地数据，每个节点都需要安装一个Flume工具，用来做数据采集。
2.2.1下载并安装
到官网去下载最新版本的Flume
下载地址为：http://flume.apache.org/，目前最新版本为1.6.0，需要1.7及以上版本的JDK。
1、解压
tar -xzvf apache-flume-1.6.0-bin.tar.gz -C /usr/local/
2、安装JDK1.7
如果节点上JDK版本低于1.7，需要安装1.7或以上版本的JDK
JDK 1.7 下载地址：
http://www.oracle.com/technetwork/java/javase/downloads/jdk7-downloads-1880260.html
在Flume目录下创建一个java目录，存放JDK
cd /usr/local/apache-flume-1.6.0-bin
mkdir java
cd java
tar -xzvf jdk-7u79-linux-x64.tar.gz
2.2.2配置Flume系统参数
修改 flume-env.sh 配置文件，主要是JAVA_HOME变量设置
cd /usr/local/apache-flume-1.6.0-bin/conf
cpflume-env.sh.templateflume-env.sh
在flume-env.sh里面设置FLUME_CLASSPATH变量和JAVA_HOME变量，
示例：
export JAVA_HOME=/usr/local/apache-flume-1.6.0-bin/java/jdk1.7.0_79
FLUME_CLASSPATH="/usr/local/apache-flume-1.6.0-bin/"
变量具体内容根据实际修改
2.2.3添加Flume第三方依赖
添加第三方依赖包flume-plugins-1.0-SNAPSHOT.jar，此包实现了一个Flume拦截器，将Flume采集到的数据进行序列化、结构化等预处理，最后每条数据生成一条Event数据返回。
cd /usr/local/apache-flume-1.6.0-bin
mkdir plugins.d --创建依赖目录，目录名必须为plugins.d
cd plugins.d
mkdir flume-plugins --项目目录，目录名随意
cd flume-plugins
mkdir lib --jar包目录，目录名必须为lib
将第三方jar包flume-plugins-1.0-SNAPSHOT.jar放在lib目录下
2.2.4添加Hive配置文件
将hive-site.xml文件拷贝到/usr/local/apache-flume-1.6.0-bin/conf目录下，并修改hive元数据地址与真实地址对应。如下所示：
<property>
<name>hive.metastore.uris</name>
<value>thrift://m103:9083,thrift://m105:9083</value>
</property>
2.2.5创建Flume agent配置文件
创建flume启动配置文件，指定source，channel，sink 3个组件内容。每个组件都有好几种配置选项，具体配置请查看Flume官网。创建配置文件flume.conf，示例如下：
vim flume.conf
a1.sources = x1
a1.sinks = y1
a1.channels = z1
# Describe/configure the source
a1.sources.x1.type = exec
a1.sources.x1.channels = z1
a1.sources.x1.command = tail -F /home/xdf/exec.txt
# Describe the sink
a1.sinks.y1.type = logger
# Use a channel which buffers events in memory
a1.channels.z1.type = memory
a1.channels.z1.capacity = 1000
a1.channels.z1.transactionCapacity = 100
# Bind the source and sink to the channel
a1.sources.x1.channels = z1
a1.sinks.y1.channel = z1
2.2.6启动Flume Agent
生产环境下，参数-Dflume.root.logger=INFO,console去掉console，此处只为方便查看测试结果，选择将日志打印到控制台。若Flume agent正常启动，说明Flume安装成功。
cd /usr/local/apache-flume-1.6.0-bin
./bin/flume-ng agent --conf conf --conf-file flume.conf --name a3 -Dflume.root.logger=INFO,console
2.2.7 测试
上面配置的example.conf文件，实现的功能是监控文件/home/xdf/exec.txt，如果有新数据写入时，Flume就会采集到新数据并打印在控制台上。
测试用例：向/home/xdf/exec.txt文件中写入内容“hello flume”，查看控制台是否打印出“hello flume”。正常输出如下：
echo 'hello flume' >> /home/xdf/exec.txt
2015-06-30 16:01:52,910 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.LoggerSink.process(LoggerSink.java:94)] Event: { headers:{} body: 68 65 6C 6C 6F 20 66 6C 75 6D 65 hello flume }
至此，Flume安装部署完毕。
3. 数据预处理
1、Flume采集数据都是按行分割的，一行代表一条记录。如果原始数据不符合要求，需要对数据进行预处理。示例如下:
原始数据格式为:
out: === START OF INFORMATION SECTION ===
out: Vendor: TOSHIBA
out: Product: MBF2300RC
out: Revision: 0109
out: User Capacity: 300,000,000,000 bytes [300 GB]
out: Logical block size: 512 bytes
经过预处理，我们将数据变为一条5个字段的记录：
TOSHIBA;MBF2300RC;0109;300;512
2、如果要将上面数据接入到hive中，我们还需要下面几个处理：
a. 创建一张hive表
create table test(Vendor string,Product string,Revision string,User_Capacity string,block string);
b. 在Kafka节点上创建一个topic，名字与上面hive表名对应，格式为“hive-数据库名-表名”。示例如下：
bin/kafka-topics --create --zookeeper localhost:2181/kafka --topic hive-xdf-test --partitions 1 --replication-factor 1
c. 将第一步得到的记录数据与topic整合成一条记录，用“@@”分割。示例如下：
hive-xdf-test @@TOSHIBA;MBF2300RC;0109;300;512
d. Flume采集整合后的一条数据，通过topic获取hive表的元数据，根据元数据对记录数据进行结构化、序列化处理，然后经过Kafka存入到hive表中。具体操作参考下面具体步骤所示。
4. Flume采集数据到Kafka
Flume如果要将采集到的数据发送到Kafka，需要指定配置文件（如下：flume_test.conf）的sink类型为KafkaSink，并且指定Kafka 的broker list。配置文件示例如下，红色标注的为KafkaSink配置项：
vim flume_test.conf
a3.channels = c3
a3.sources = r3
a3.sinks = k3
a3.sources.r3.type = exec
a3.sources.r3.channels = c3
a3.sources.r3.command = tail -F /home/xdf/exec.txt
a3.sources.r3.fileHeader = false
a3.sources.r3.basenameHeader = false
a3.sources.r3.interceptors = i3
a3.sources.r3.interceptors.i3.type =iie.flume.interceptor.CSVInterceptor$Builder
a3.sources.r3.interceptors.i3.separator = ;
a3.sources.r3.decodeErrorPolicy=IGNORE
a3.channels.c3.type = memory
a3.channels.c3.capacity = 10000
a3.channels.c3.transactionCapacity = 1000
a3.sinks.k3.channel = c3
# a3.sinks.k3.type = logger
#a3.sinks.k3.batchSize = 10
a3.sinks.k3.type = org.apache.flume.sink.kafka.KafkaSink
a3.sinks.k3.brokerList = localhost:9092
注意：此处有一个拦截器插件的定义，它就是用来做结构化、序列化数据预处理的。此插件由上面配置的Flume第三方jar包中获得。
a3.sources.r3.interceptors.i3.type =iie.flume.interceptor.CSVInterceptor$Builder
5. Kafka数据落地
我们提供了一个Camus工具，来定时将Kafka中的数据落地到hive表中。
Camus工具包含以下三个文件：
|
文件 |
说明 |
|
camus-example-0.1.0-cdh-SNAPSHOT-shaded.jar |
程序运行jar包 |
|
camus.properties |
配置文件 |
|
camusrun.sh |
运行脚本 |
配置文件需要根据实际情况，修改以下两个参数
kafka.whitelist.topics=hive-xdf-test ----数据对应的topic
kafka.brokers=m105:9092,m103:9092 ----kafka broker lists
需要指定多个topic时，用逗号间隔，示例：
Kafka.whitelist.topics=topic1,topic2,topic3
修改完配置文件后，定时运行camusrun.sh脚本，就会将新生成的数据接入到topic所对应的hive表中了。
6. 具体案例
6.1 Smart数据接入
6.1.2 创建hive表
最终我们要将smart数据接入到hive表中，所以我们首先要创建一个满足smart数据结构的hive表。
create table smart_data(serial_number String ,update_time string,smart_health_status string ,current_drive_temperature int,drive_trip_temperature int,elements_in_grown_defect_list int,manufactured_time string ,cycle_count int ,start_stop_cycles int ,load_unload_count int ,load_unload_cycles int ,blocks_sent_to_initiator bigint ,blocks_received_from_initiator bigint ,blocks_read_from_cache bigint ,num_commands_size_not_larger_than_segment_size bigint ,num_commands_size_larger_than_segment_size bigint ,num_hours_powered_up string ,num_minutes_next_test int ,read_corrected_ecc_fast bigint ,read_corrected_ecc_delayed bigint ,read_corrected_re bigint ,read_total_errors_corrected bigint ,read_correction_algo_invocations bigint ,read_gigabytes_processed bigint ,read_total_uncorrected_errors string ,write_corrected_ecc_fast bigint ,write_corrected_ecc_delayed bigint ,write_corrected_re bigint ,write_total_errors_corrected bigint ,write_correction_algo_invocations bigint ,write_gigabytes_processed bigint ,write_total_uncorrected_errors string ,verify_corrected_ecc_fast bigint ,verify_corrected_ecc_delayed bigint ,verify_corrected_re bigint ,verify_total_errors_corrected bigint ,verify_correction_algo_invocations bigint ,verify_gigabytes_processed bigint ,verify_total_uncorrected_errors bigint ,non_medium_error_count bigint);
6.1.2 创建topic
Flume采集到的数据要生成一条条的event数据传给kafka消息系统保存，kafka需要事先创建一个topic来生产和消费指定数据。为系统正常运行，我们统一定义topic的名字结构为“hive-数据库名-表名”。需要在kafka集群节点上创建topic，示例如下：
bin/kafka-topics --create --zookeeper localhost:2181/kafka --topic hive-xdf-smart_data --partitions 1
--replication-factor 1
注意：此处的数据库名、表名，必须为上一步创建的hive表，因为Flume会通过此topic名来获取hive表的元数据信息，从而生成对应event数据。
6.1.2 配置Flume agent启动参数
生成参数文件smart_test.conf如下：
vim smart_test.conf
a3.channels = c3
a3.sources = r3
a3.sinks = k3
a3.sources.r3.type = exec
a3.sources.r3.channels = c3
a3.sources.r3.command = tail -F /home/xdf/exec.txt
a3.sources.r3.fileHeader = false
a3.sources.r3.basenameHeader = false
a3.sources.r3.interceptors = i3
a3.sources.r3.interceptors.i3.type =iie.flume.interceptor.CSVInterceptor$Builder
a3.sources.r3.interceptors.i3.separator = ;
a3.sources.r3.decodeErrorPolicy=IGNORE
a3.channels.c3.type = memory
a3.channels.c3.capacity = 10000
a3.channels.c3.transactionCapacity = 1000
a3.sinks.k3.channel = c3
# a3.sinks.k3.type = logger
#a3.sinks.k3.batchSize = 10
a3.sinks.k3.type = org.apache.flume.sink.kafka.KafkaSink
a3.sinks.k3.brokerList = localhost:9092
注意：
1、此处数据源sources的类型为exec。具体命令为：
a3.sources.r3.command = tail -F /home/xdf/exec.txt
我们定时在每个节点运行一个脚本生成一条smart数据，将数据写入/home/xdf/exec.txt文件。
flume用上面那个命令一直监控文件/home/xdf/exec.txt，如有新数据写入，则采集传输到kafka里。
2、指定了一个自定义的第三方插件，Flume过滤器CSVInterceptor，将CSV格式的数据转化成结构化，序列化的Event格式。
3、Sink为KafkaSink，数据会写到kafka里面，特别注意：这里需要指定对应的brokerList，示例如下：
a3.sinks.k3.brokerList = m103:9092,m105:9092
6.1.3 开启Flume Agent
执行命令：
cd /usr/local/apache-flume-1.6.0-bin
./bin/flume-ng agent --conf conf --conf-file smart_test.conf --name a3 -Dflume.root.logger=INFO
6.1.4 生成Smart数据
在每个数据节点上运行createEvent.py脚本，生成一条结构化好的smart数据。
脚本有两个参数smart_data.log，hive-xdf-smart_data，前者为smart命令输出的原始信息文件，后者是topic名字，即上一步生成的topic名。
python createEvent.py smart_data.log hive-xdf-smart_data >
/home/xdf/exec.txt
此脚本会解析smart原始信息，生成一条带topic字段的结构化smart数据写入到/home/xdf/exec.txt文件中，数据格式如下：
hive-xdf-smart_data@@EB00PC208HFC;2015-06-23 18:56:09;OK;28;65;0;week 08 of year 2012;50000;21;200000;69;-1;-1;-1;-1;-1;-1;-1;0;0;0;0;0;0;300744.962;0;0;0;0;0;0;10841.446;0;-1;-1;-1;-1;-1;-1;-1
用符号“@@”将topic跟smart数据分开，smart数据每列间用逗号隔开。
6.1.5 测试时查看Kafka数据
查看数据是否成功生成到kafka中，可在kafka节点上，通过下面命令查看：
kafka-console-consumer --zookeeper localhost:2181/kafka --topic hive-xdf-smart_data --from-beginning
结果展示：
6.1.6 Kafka数据落地到hive表中
打开camus.properties配置文件，修改以下两个参数
kafka.whitelist.topics=hive-xdf-smart_data ----smart数据对应topic
kafka.brokers=m105:9092,m103:9092 ----kafka broker lists
修改完配置文件后，定时运行camusrun.sh脚本，就会将新生成的smart数据接入到topic所对应的hive表中了。
至此，数据接入流程完毕。