一,目的
在学习的过程中,需要用到 PDI ---一个开源的ETL软件。主要是用它来设计一些转换流程来处理数据。但是,在PDI中设计好的 transformation 是在本地的执行引擎中执行的,(参考源码中的 Trans.java ),现可以对DI加以改造:在DI中设计的转换,将之转换成Storm的Topology,然后再把该Topology提交到Storm集群中执行。这样,既可以利用DI强大的设计能力(因为在DI中可以设计各种各样的转换流程,这些用DI设计出来的 transformation流程是前人已经实现好的数据处理功能,把该 transformation 转换成 Storm Topology 可以避免自己编写实现数据处理功能的Storm Topology 代码);又可以利用Storm的分布式实时流处理数据的能力。
在Pentaho Lab 官网上有一个相应的开源项目,但是貌似已经不再更新了。参考: kettle-storm 的 github 以及 相关介绍
二,实现概述
①Storm 端的运行流程:
不管在DI中设计了何种转换,将转换 转化成 Storm 的Topology时,都只得到两种类型的Topology,即输入拓扑和处理拓扑。提交Topology的代码如下:
1 StormSubmitter.submitTopology(basictopo.getName(), config, basictopo.getInputTopology());//commit inputtopology 2 StormSubmitter.submitTopology(basictopo.getName(), config, basictopo.getProcessTopology());//commit processtopology
②PDI中的转换设计
一个典型的转换如下图:图片来源
设计好一个这样的转换流程图后,在运行该转换前,需要将设计好的转换保存成一个 .ktr 文件,PDI从该.ktr 文件中解析出 transMeta 对象,(transMeta对象 代表运行过程中的 某个转换)。transMeta对象记录了整个转换过程中需要的所有信息。而我们的目标就是将上图中的设计的transformation 自动 变成 Storm的Topology。从而,可以在Storm中执行如上图所示的 transformation了。
③transformation 到 Topology 的变换
由上图设计的转换可以看出,它是一个有向无环图。而Storm的Topology也是一个DAG图,因此可以通过算法将 transformation 变成 Topology。
三,细节及遇到的问题
❶transMeta 对象不能直接序列化,如何在Storm集群中获得transMeta对象,然后根据该对象构造Topology?
主要用 .ktr 文件 来生成 TransMeta对象。当在本地运行时,TransMeta对象代表整个转换的执行,但当提交到集群中时,本地执行过程中的transMeta对象不可能直接提交到集群中去的。需要将之序列化,再发送到集群中。而,每个 .ktr文件就代表了一个转换,因此可以直接将 .ktr 文件打包上传到集群中,然后再解析jar包中的 .ktr 文件来生成 构成 InputSpout 和 Bolt 所需的transMeta 对象。关于序列化知识,可参考:
InputSpout 解析transMeta对象:
URL fileURL = this.getClass().getResource("/test-random.ktr"); ktrFile = fileURL.toString(); transMeta = new TransMeta(ktrFile, (Repository) null);
TransactionalBolt解析transMeta对象:
InputStream inputStream = this.getClass().getResourceAsStream("/test-random.ktr"); if(transMeta == null ) { try{ KettleEnvironment.init(); transMeta = new TransMeta(inputStream, (Repository)null, true, null, null); }catch(Exception e){ e.printStackTrace(); e.getMessage(); } }
遇到了一个奇怪的问题:即InputSpout中解析transMeta对象是在本地完成的,也即在Topology的main函数所在的类执行时完成的。而TransactionalBolt.java中解析transMeta对象是在将Topology提交到集群上之后,在集群上执行时去读取 .ktr 文件并解析transMeta对象。
因此,InputSpout.java中读取 .ktr 文件解析transMeta对象时,是按照文件流的形式读取的。而在把Topology提交到集群后, .ktr 文件是随着Topology的jar包一起打包提交的,因此读取jar包中的 .ktr文件。
❷Storm执行Topology时需要依赖DI中的jar包,依赖包问题如何解决?
将DI的执行引擎换成了Storm执行引擎,但不管是什么执行引擎,执行的代码还是会依赖PDI的lib目录下的一些依赖包。比如,PDI连接数据库时需要一个 mysql-connector-java-5.1.26.jar ,当把从 transformation中生成的Topology提交到Storm集群时,若transformation中有访问数据库的操作且storm 安装目录下的 lib 目录下没有相应的jar包时,Topology会执行失败。因此需要将mysql-connector-java-5.1.26.jar 复制到 storm 安装目录下的 lib 目录中。同理,其他的一些依赖jar包都需要复制到 storm 安装目录下的 lib 目录中。依赖的包如下:
metrics-core-2.2.0.jar
scala-library-2.11.5.jar
kafka_2.11-0.8.2.0.jar
esapi-2.0.1.jar
source.jar
etl-core-5.0.jar
kafka-clients-0.8.2.0.jar
mysql-connector-java-5.1.26.jar
zookeeper-3.4.6.jar
commons-vfs-20100924-pentaho.jar
❸Topology的构造需要依赖transMeta对象,进而需要依赖DI源码中的 .class 文件,如何正确打包 storm.jar 并自动提交(不是通过命令行 运行 storm jar 进行提交)?
命令行的storm jar 命令是Python 脚本实现的,它完成的功能主要是设置环境变量(将jar文件的路径设置成 "storm.jar" 环境变量STORM_JAR)并打包成jar文件,该部分功能可以在主类中实现,从而不需要在命令行运行 storm jar 命令进行Topology的提交。
在具体的学习中,直接把整个工程的 bin 目录下的所有编译好的文件都打进jar包,这样可以解决Topology运行时依赖DI中的代码而报的 java.lang.NoClassDefFoundError 错误。但是,这样做使得需要提交的jar包非常的大!
四,缺点
由于PDI中转换中的许多插件都不是针对分布式环境而设计的,因此,在PDI中设计的部分 transformation 是不能正常运行在 Storm之上的,可参考Pentaho Lab中关于Kettle On Storm的描述。
其次,PDI中已有的一些插件不能满足大数据集群的需求,需要重新开发新的PDI插件。