storm-安装

Posted 2020-09-05 woxiaohaha

1. 单机安装storm测试环境

http://www.myexception.cn/open-source/1809563.html

 

2. storm 将处理结果保存到mysql

  1)下载mysql jar http://www.mysql.com/downloads/connector/j/

  2) 将jar包上传到storm lib目录

  3) 将jar 导入到idea 

  4) 按照java 操作mysql 即可

 

需求，将storm的一个数据流，直接保存到es，可以省略很多环节!

 

测试1： 百度到 公认的使用 storm-elastic-search 一个jar 很是方便，在maven中进行如下配置

             <dependency>

              <groupId>com.hmsonline</groupId>

 <artifactId>storm-elastic-search</artifactId>
 <version>0.1.0</version>
 </dependency>

         同时，部署es 下载最新的2.4.0，安装

         wget https://download.elastic.co/elasticsearch/release/org/elasticsearch/distribution/tar/elasticsearch/2.4.0/elasticsearch-2.4.0.tar.gz

         tar zxvf elasticsearch-2.4.0.tar.gz

         cd elasticsearch-2.4.0

         安装es plugin插件

         bin/plugin install mobz/elasticsearch-head

        访问http://localhost:9200

        问题1: 在storm上进行测试，提示找不到某些class,需要将storm-elastic-search相关的jar包，都上传到storm安装目录lib下。

        问题2: 运行后 错误提示 没有节点可用，然后在es错误文件显示，连接库版本为0.90，而当前需要的版本最少为2.0 版本不兼容...

        下载一个es0.90版本，发现可以使用

        接着安装kibana 4 ，发现不能与es0.9兼容，支持最低es版本为1.4.4

        升级es0.9到es1.4.4后，发现storm-elastic-search不能与es1.4.4兼容。

测试2：搜索其它storm es jar包，发现https://github.com/apache/storm

            其中 样例代码 https://github.com/apache/storm/tree/master/external/storm-elasticsearch

            看一下代码 最重要的是EsIndexBolt, 因为需求只是将数据上传到es中，为了与最新es2.4兼容，去掉了storm-elastic-search的依赖，将es2.4安装文件的lib文件统一copy到storm安装目录下的lib中，开始了繁琐的代码修改。因为EsIndexBolt是基于旧版本的es0.90编写的。

          每个文件中的org.apache.storm的引用都需要去掉，变成backtype.storm的引用。

          EsConfig 中ImmutableSettings.settingsBuilder()   需要修改成Settings.Builder(), 因为新版es jar 没有ImutableSettings

       DefaultEsTupleMapper.java 将source 修改为document, 这个是自定义的，与自己的bolt相同即可

         StormElasticSearchClient.java 38行 TransportClient transportClient = new TransportClient(settings); 新的es  TransportClient 没有此构造方法需要修改为

     TransportClient transportClient = TransportClient.builder().settings(settings).build();

         TransportAddresses.java 中的 transformToInetAddress 方法，修改为

return new InetSocketTransportAddress(InetAddress.getByName(hostname), port(hostAndPort[1])); 因为新版本es

InetSocketTransportAddress 类的构造函数发生变化