0


一百零八、Kettle采集Kafka数据到HDFS(踩坑,亲测有效)

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中就全部结束了,坑也挺多的。

乐于奉献共享,帮我你我他!!!

标签: hdfs kafka hadoop

本文转载自: https://blog.csdn.net/tiantang2renjian/article/details/130576805
版权归原作者 天地风雷水火山泽 所有, 如有侵权,请联系我们删除。

“一百零八、Kettle采集Kafka数据到HDFS(踩坑,亲测有效)”的评论:

还没有评论