背景
linux下使用kettle（pentaho data integration）将mysql数据，通过 Parquet Output 导入CDH 6.2集群hive的parquet表中。
解决了直接用 Parquet Output 导入hive parquet 表中字段不匹配的问题。
步骤：
1. 下载 pdi 8.3，解压后赋予相应的权限
unzip pdi-ce-8.3.0.0-371.zip
cd data-integration/
chmod +x -R *.sh
在vnc中，data-integration目录执行 ./spoon.sh，打开操作界面。
2. 新建转换，选择“主对象树”，配置DB连接（mysql连接）
填写对应的mysql连接信息。点击测试显示正确连接。
如果此处报错缺少msyql驱动，则把 mssql-jdbc-7.0.0.jre10.jar 拷贝到 data-integration/lib 下，注意 .jar 权限，重启 pdi 有效。
3. 编辑转换
选择“核心对象”，从“输入”中拖出“表输入”，双击打开
选择上一步创建的mysql连接，并填写sql语句。可以点击预览查看数据。
4. 连接到 hive 输出
4.1 更新pdi中CDH集群的配置文件
从 "Cloudera Manager -- HDFS -- 下载客户端配置" 下载 core-site.xml 和 hdfs-site.xml
更新到 pdi 目录下的 data-integration\plugins\pentaho-big-data-plugin\hadoop-configurations\cdh61
打开 core-site.xml 和 hdfs-site.xml，确认所有的 url 都是对应的 hostname 而不是 localhost
4.2 连接CDH集群
在 “主对象树 -- Hadoop clusters” 下新建 cluster，填写对应信息
注意：此处所有 hostname 不能写 localhost，要改为对应的 hostname
点击测试，查看连接
可以看到 HDFS 已经成功连接。
4.3 新建 Parquet Output
在 “核心对象 -- Big Data” 下拖出 Parquet Output，按住 shift 单击 "mysql输入"，拖出连接线连到 "Parquet Output"，双击打开
点击“浏览”，选择之前配置好的 cluster，选择目标地址。
这里的地址是到文件名的，但不包含后缀，后续导入到hive中时会自动添加后缀名
所以需要在选好的地址后手动在地址后面添加 /<filename>
注意：这里有个BUG，就是不能将数据直接导在已有的表中，会报字段类型不匹配的问题。因为hive中已有表的字段类型（比如 INT）和parquet表的字段类型（比如 Int64，见下图）不一致。
直接导入表，刷新数据后查询的报错
[slave01:21000] test> select * from log_operation;
Query: select * from log_operation
Query submitted at: 2019-11-25 15:01:38 (Coordinator: http://slave01:25000)
Query progress can be monitored at: http://slave01:25000/query_plan?query_id=af4f332a5f30f746:7f0a5b4900000000
ERROR: File 'hdfs://master:8020/user/hive/warehouse/test.db/log_operation/f.parquet' has an incompatible Parquet schema for column 'test.log_operation.instance_id'. Column type: BIGINT, Parquet schema:
optional byte_array instance_id [i:1 d:1 r:0]
对于这个问题，笔者正巧因工作原因咨询过Pentaho负责开发pdi的工作人员，得到答案可通过将源表里的数据都转为String暂时克服。不过笔者通过另一种方式绕过了这个BUG，下面介绍这个曲线救国方法。
笔者通过先将表导入一个临时的文件夹内（如上一步选路径所示，选数据库下的临时文件夹xxx.db/tmp。实际是选到数据库，手动添加后面的路径），然后通过sql（impala sql 或 hive sql）根据导入的文件创建表，然后再把/tmp下的文件加载过来，就完成了，顺便绕过了上面所说字段不一致的BUG。
下面继续
选好地址后，回到 Parquet Output 界面，点击Get Fields自动获取字段（这里可以看到parquet的数据格式）。然后确定。
（定时任务的话，可以通过Option那个tab下勾选“文件名包含时间”等选项避免每次导入的文件因重名而覆盖）
5. 新建 “执行sql语句”
由于笔者用的是impala，所以本步骤是在讲连接到impala执行sql。使用hive的话相应步骤改为hive就可以，是一样的。
5.1 连接impala
将 impala jdbc 驱动 ImpalaJDBC41.jar 拷贝到 data-integration/lib 和 data-integration/plugins/pentaho-big-data-plugin/lib 和 data-integration/plugins/pentaho-big-data-plugin/hadoop-configurations/cdh61/lib 下。注意权限，重启有效。
按照步骤2，选择“主对象树”，配置DB连接，连接到 impala
5.2 新建 “执行sql脚本”
拖出“核心对象” -- “脚本” -- “执行sql脚本”
双击打开，填入sql脚本
create table log_operation
like parquetfile '/user/hive/warehouse/test.db/tmp/init.parquet'
stored as parquet
location '/user/hive/warehouse/test.db/log_operation';
load data inpath '/user/hive/warehouse/test.db/tmp/init.parquet'
into table log_operation;
refresh log_operation;
三个命令分别是：
- 按照导入文件（
test.db/tmp/init.parquet）的格式创建表 - 将文件导入到创建的表中
- 刷新表
这样建的表和parquet表的字段格式是匹配的，即使后续继续往表中直接导入增量数据（同样通过 parquet output），也不会出现字段格式不匹配的错误。
5.3 加入“停止等待”
pdi 转换的 执行sql语句 虽然连接在 Parquet Output 之后，但确是并行执行的。因此需要加入停止等待，等待 Parquet Output 执行完后，再执行sql。
至此，可以直接点击转换下的运行图标运行。
也可以打包成作业，方便调度。
6. 新建作业
保存之前的转换。新建作业。
从“核心对象”拖出 start 和 转换，双击转换配置，添加刚刚保存的转换。确定。
执行作业。
作为job，可以在 Start 里设置调度时间，然后运行就可以了。
网友评论