Storm Topology 提交 总结---Kettle On Storm 实现

一，目的

在学习的过程中，需要用到 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的代码如下：

StormSubmitter.submitTopology(basictopo.getName(), config, basictopo.getInputTopology());//commit inputtopology
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插件。