温馨提示：要看高清无码套图，请使用手机打开并单击图片放大查看。
Fayson的github：
https://github.com/fayson/cdhproject
提示：代码块部分可以左右滑动查看噢
1.文档编写目的
内容概述
1.测试环境准备
2.配置StreamSets
3.创建Pipline及测试
4.总结
测试环境
1.RedHat7.3
2.CM和CDH版本为cdh5.13.3
3.Kafka2.2.0(0.10.2)
4.StreamSets3.3.0
前置条件
1.集群已启用Kerberos
2.集群已安装Kafka并启用Kerberos
2.测试环境准备
1.准备一个访问Kerberos环境的Kafka的fayson.keytab文件
[root@cdh01 ~]# kadmin.local
kadmin.local: listprincs fayson*
fayson@FAYSON.COM
kadmin.local: xst -norandkey -k fayson.keytab fayson@FAYSON.COM
(可左右滑动)
fayson.keytab主要在向Kafka生产消息和StreamSets消费Kafka数据时使用。
2.准备向Kerberos环境的Kafka集群生产数据脚本
该脚本用于向Kafka发送JSON数据，脚本说明：
run.sh：向Kafka指定topic生产数据的脚本
ods_user_600.txt：发送到Kafka的测试数据，共600条测试数据，数据的id是唯一的。
注意：发送数据的示例代码是将ods_user_600.txt的每条数据转换为json格式了，示例数据如下：
{
"occupation": "生产工作、运输工作和部分体力劳动者",
"address": "台东东二路16号-8-8",
"city": "长治",
"marriage": "1",
"sex": "1",
"name": "仲淑兰",
"mobile_phone_num": "13607268580",
"bank_name": "广州银行31",
"id": "510105197906185179",
"child_num": "1",
"fix_phone_num": "15004170180"
}
(可左右滑动)
lib：Fayson打包的示例jar包及Kafka的依赖包
conf：示例代码运行的配置文件
该脚本运行主要依赖0286.properties、jaas.conf、krb5.conf和fayson.keytab文件。
0286.properties配置文件，将如下配置修改为自己集群的环境：
[root@cdh04 conf]# vim 0286.properties
bootstrap.servers=cdh02.fayson.com:9092,cdh03.fayson.com:9092,cdh04.fayson.com:9092
topic.name=kafka_hive_topic
krb5.debug=false
group.id=testgroup
(可左右滑动)
jaas.conf文件内容如下：
[root@cdh04 conf]# vim jaas.conf
KafkaClient {
com.sun.security.auth.module.Krb5LoginModule required
useKeyTab=true
keyTab="/data/disk1/0286-kafka-shell/conf/fayson.keytab"
principal="fayson@FAYSON.COM";
};
Client {
com.sun.security.auth.module.Krb5LoginModule required
useKeyTab=true
storeKey=true
keyTab="/data/disk1/0286-kafka-shell/conf/fayson.keytab"
principal="fayson@FAYSON.COM";
};
(可左右滑动)
krb5.conf：拷贝集群的krb5.conf文件到conf目录下即可。
3.通过如下命令创建测试topic
[root@cdh04 conf]# kafka-topics --create --zookeeper cdh01.fayson.com:2181 --replication-factor 3 --partitions 1 --topic kafka_hive_topic
(可左右滑动)
4.通过Hue为sdc用户授权
授予default库的所有权限以及/user/hive/warehouse目录的URI权限，否则sdc用户无法创建表。
3.StreamSets配置
由于Kafka集群启用了Kerberos，所以这里在使用StreamSets消费Kafka数据之前，需要配置StreamSets访问Kafka的Kerberos用户信息，具体配置如下：
1.登录Cloudera Manager并进入StreamSets服务
2.点击“配置”，搜索“sdc_java_opts”，在该配置项中增加如下内容
-Djava.security.auth.login.config=/data/disk1/0286-kafka-shell/conf/jaas.conf
(可左右滑动)
注意：jaas.conf文件需要存在于StreamSets的Data Collector服务所在节点的指定目录下。
4.创建StreamSets的Pipline
1.登录StreamSets，创建一个kafka2kudu的Pipline
2.在Pipline流程中添加Kafka Consumer作为源并配置Kafka基础信息
注意：Kafka的版本选择，这里Fayson选择的Apache的Kafka，在Kerberos环境下选择CDH对应的Kafka版本会报“Couldnot get partition count for topic 'kafka_hive_topic'”
配置Kafka相关信息，如Broker、ZK、Group、Topic及Kerberos信息
注意：访问Kerberos环境的Kafka需要配置security.protocol=SASL_PLAINTEXT和sasl.kerberos.service.name=kafka两个参数。
配置数据格式化方式，写入Kafka的数据为JSON格式，所以这里选择JSON
3.添加Hive Metadata中间处理模块，选择对应的CDH版本
配置Hive的JDBC信息
注意：这里访问Hive的JDBC连接，需要添加Kerberos信息，由于无法通过StreamSets界面指定我们自定义用户访问Hive，所以这里默认使用的是StreamSets的sdc用户，如果集群启用了Sentry则需要注意为sdc用户授权，否则无法创建hive表和写数据。
配置Hive的表信息，指定表名和库名
指定数据格式，指定为Avro，选项中有parquet格式，但在后续处理中并不支持parquet格式
4.添加Hadoop FS处理模块，主要用于将HiveMetadata的数据写入HDFS
配置Hadoop FS，配置HDFS URL和是否启用Kerberos认证
配置Hadoop FS的Out Files
注意：勾选“Directory in Header”使HDFS写入数据时使用上一步中Hive Metadata模块传递的目录，“IdleTimeout”主要是用于指定Hadoop FS模块空闲多久则将数据刷到HDFS数据目录。
配置Late Records参数，使用默认参数即可
指定写入到HDFS的数据格式
5.添加Hive Metastore模块，该模块主要用于向Hive库中创建表
配置Hive信息，JDBC访问URL
Hive Metastore的高级配置
6.点击校验流程，如下图所示则说明流程正常
到此为止完成了Kafka数据到Hive的流程配置。
5.流程测试验证
1.启动kafka2hive的Pipline，启动成功如下图显示
2.在命令行运行run.sh脚本向Kafka发送消息
[root@cdh04 ~]# cd /data/disk1/0286-kafka-shell/
[root@cdh04 0286-kafka-shell]# sh run.sh ods_user_600.txt
(可左右滑动)
3.在StreamSets中查看kafka2hive的pipline运行情况
4.使用sdc用户登录Hue查看ods_user表数据
入库的数据总条数
可以看到ods_user表的总条数与准备的测试数据量一致。
6.总结
1.Kafka集群启用了Kerberos后，StreamSets的Kafka模块在消费数据时需要在sdc_java_opt中加载jaas.conf，指定消费Kafka数据的Kerberos账号。
2.Hive Metadata模块主要是用于将Kafka的JSON数据进行封装分流处理，data数据交给HDFS模块，MetaData数据交个HiveMetastore模块，HDFS模块主要用于写数据到hive表的数据目录，HiveMetastore主要用于判断表是否存在是否需要创建表。
3.由于HiveMetastore模块无法指定自定义的Kerberos账号，默认使用sdc用户访问Hive，在启用Sentry的集群则需要为sdc用户授权，否则无权限创建表。
4.HDFS模块在接收到HiveMetadata模块的数据后生成的为临时文件，不是立即将数据写入到HDFS，可以通过“Idle Timeout”参数来控制刷新数据到HDFS的频率。
GitHub地址：
生产消息脚本地址：
https://github.com/fayson/cdhproject/tree/master/kafkademo/0286-kafka-shell
源码地址：
https://github.com/fayson/cdhproject/blob/master/kafkademo/src/main/java/com/cloudera/kerberos/ReadUserInfoFileToKafka_0286.java
提示：代码块部分可以左右滑动查看噢
为天地立心，为生民立命，为往圣继绝学，为万世开太平。
温馨提示：要看高清无码套图，请使用手机打开并单击图片放大查看。
推荐关注Hadoop实操，第一时间，分享更多Hadoop干货，欢迎转发和分享。
原创文章，欢迎转载，转载请注明：转载自微信公众号Hadoop实操