Kafka到HDFS,除了用Kafka API和flume之外,还可以用kettle,最大优点是不用写代码!
版本:Kettle版本:8.2、Hadoop版本:3.1.3
前提: 详情请看鄙人的一百零一、Kettle8.2.0连接Hive3.1.2(踩坑,亲测有效)
http://t.csdn.cn/mWfOChttp://t.csdn.cn/mWfOC
前提一、Hadoop系列配置文件已复制到kettle路径下 路径为:D:\java\kettle\pdi-ce-8.2.0.0-342\data-integration\plugins\pentaho-big-data-plugin\hadoop-configurations\hdp30
注意:以上的配置文件要与自己使用的服务器配置文件一致
前提二、在D:\java\kettle\pdi-ce-8.2.0.0-342\data-integration\plugins\pentaho-big-data-plugin\hadoop-configurations\hdp30\lib文件夹下注意替换jar包
前提三、在D:\java\kettle\pdi-ce-8.2.0.0-342\data-integration\lib增加MySQL驱动包,注意驱动包版本问题
这些准备好之后,下面开始用kettle采集Kafka中的数据到hdfs!
第1步,在kettle中创建新的转换任务。
第2步,在Streaming中拖拽Kafka consumer控件,并且修改Kafka consumer控件配置信息
首先,Setup模块
**
Transformation
:最好新建一个,用于返回流中的结果**
**
Connection
:选择direct ,右侧输入bootstrap-server IP地址:端口号9092**
**
Topics
:选择kafka消费的topic**
**
Consumer group
:消费者组,随便填**
其次,Batch模块(根据自己的实际情况修改参数)
Duration: 处理批次间隔
Number-of-records: 处理批次条数
maximum concurrent batches:最大并发批次(8.2版本没有,9.3版本有)
message prefetch limit:消息预取限制,防止读取数据量太大,造成kettle挂掉(8.2版本没有,9.3版本有)
Offset-management: 偏移量管理
最后,Result fields模块
其余,Fields和Options两个模块不需要修改。
第3步,拖拽一个应用模块里面的写日志控件,方便查看Kafka中的字段
第4步,拖拽一个转换模块里面的字段选择控件,只取Kafka日志中的Message字段
第5步,是关键的一步!!! 拖拽输入模块里的JSON input插件
首先,在文件模块页面,勾选源定义在一个字段里,并从Message字段获取源。可以修改步骤名称
其次,在字段模块。如果Select fields查不到字段的话,别慌。手动输入即可,不过此处为第一级别的字段,二级字段不在此处。不需要很详细。另外,要注意字段的类型
这是第一个大坑。明明Kafka里有数据,JSON input控件却找不到字段,最后发现可以手动输入字段名。
第6步,由于还有二级字段需要解析,因此还要再拖拽一个JSON input字段
首先,拖拽输入模块里的JSON input插件
其次,在文件模块,勾选源定义在一个字段里,并从evaluationList字段获取源
最后,在字段页面,Select fields字段。如果找不到,手动输入即可。注意,字段类型
第7步,为了方便查看目前已有的字段,并且修改一些字段的名称,再次拖拽转换模块里面的字段选择控件
首先,在选择和修改页面,获取选择的字段,并且把不用的字段删掉
其次,在元数据页面,修改需要修改的字段名称,注意字段类型以及格式的修改,特别是时间戳类型的字段,不能用string
第8步,最后一个控件。由于我这边是从Kafka读取数据到HDFS,所以我这边是拖拽Big Data里的Hadoop file output控件
首先,拖拽Big Data里的Hadoop file output控件
其次,在文件页面。
1.选择Hadoop Cluster 如果没有,可以新建一个 (前提是开启了Hadoop)
1.1.点击New新建Hadoop Cluster
cluster name:随便填
storage:选择HDFS
HDFS的Hostname、Port,注意与Hadoop配置文件core-site.xml里填的一模一样
Username和Password就填自己的服务器密码,一般用户名是root,密码就是自己设置的root用户密码
注意:如果配置文件core-site.xml里hostname是hurys22,那么kettle里的Hostname也必须是hurys22,而不能是192.168.1.22这种,这是第二个大坑。
1.2 这部分填好即可,剩下的部分不用填。填好后测试一下
这4个好即可,其他不需要
2.在Folder/File,选择hdfs的文件路径。可以先自己创建好,也可以让它自动生成。不过,自己要先创建好目标文件夹,因为她有个文件权限的问题。这是第三个大坑!!!
其实,经踩坑后发现,目标文件evaluation.csv,kettle任务可以自行创建,不过目标文件夹rtp,如果HDFS没有则必须提前手动创建好,然后给这个文件夹赋权即可。
2.1 创建hdfs目标文件夹
[root@hurys22 soft]# hdfs dfs -mkdir -p /rtp
2.2 文件赋权
[root@hurys22 ~]# hdfs dfs -chmod -R 777 /rtp/
注意:必须要修改文件夹权限,否则kettle会报错文件权限不够
Caused by: org.apache.hadoop.ipc.RemoteException(org.apache.hadoop.security.AccessControlException): Permission denied: user=Administrator, access=WRITE, inode="/rtp":root:supergroup:drwxr-xr-x
3.如果自己已经创建好目标文件,那么要勾选 启动时不创建文件
4.文件扩展名填csv
5.一定要指定日期时间格式,一般选择年月日或者年月日时分秒。否则。你第二次导到HDFS的数据会覆盖第一次导出的数据,因为文件路径是一样的 这是第四个大坑!!!
6.一个小技巧,可以通过显示文件名查看即将生成的目标文件名,从而可以确认文件名是否正确
7.在内容页面。选择分隔符,一般是,或者; 还有编码:UTF-8
8.在字段页面,点击获取字段和最小宽度,检查所需字段以及字段类型,注意时间戳字段类型的格式
第9步,在Hadoop file output控件修改好,保存kettle任务,点击运行
同时,HDFS会在提前建好的目标文件夹下自动生成目标文件,当任务停止时,文件才会显示数据大小。
不过,可以直接下载查看
第一步,点击Download
第二步,在下载文件夹下可以看到刚才下载的文件evaluation.csv以及文件大小
第三步,用notepad++打开目标文件evaluation.csv查看数据
截止这边,用kettle采集Kafka数据到HDFS中就全部结束了,坑也挺多的。
乐于奉献共享,帮我你我他!!!
版权归原作者 天地风雷水火山泽 所有, 如有侵权,请联系我们删除。