storm 多语言支持实践

原理

• shell bolt 将数据和命令序列化成 json 后输出到shell 进程中，shell 进程接收 json 解析后将结果输出到 stdout 中，shell bolt 获取 stdout 的输出进行解析然后调用相应的 api 进行后续操作（emit，ack，fail 等）
• 相关协议见：http://storm.apache.org/releases/1.1.1/Multilang-protocol.html

实现：

• SimpleShellBolt

public class SimpleShellBolt extends ShellBolt implements IRichBolt {

    public SimpleShellBolt() {
        // 划重点：此处注意传参的方式 不要写成一个 string, 比如"sh start.sh"
        // start.sh 作用是对二进制文件 加执行权限后进行调用，否则会提示没执行权限
        super("sh", "start.sh");

    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("word"));
    }

    @Override
    public Map<String, Object> getComponentConfiguration() {
        return null;
    }
}

• shell 脚本
  • shell 脚本需要放置到 src/main/resources/resources 目录中
  • submit 时 storm 会将 jar 包中的 resources 目录拷贝到 storm 工作目录下的resources 目录下(比如：storm-local/supervisor/stormdist/kafka-2-shell-topic-consumer-38-1513754394/resources/)
  • shell bolt 运行 shell 进程时会 cd 到storm 工作目录下的resources进行启动

踩过的坑

• 使用的 storm cpp 程序中对 heart beat 的判断不准（估计是 shellbolt 代码升级所致）导致抛出『Anchored onto * after ack/fail』 的异常

  • stormcpp 源码https://github.com/honeyligo/StormCpp
  • 修改

  vim https://github.com/honeyligo/StormCpp/blob/master/public/Storm.h
  bool IsHeartbeatTuple()
        {
            // if (this->_task == -1 && this->_stream.compare("__heartbeat") == 0)
            if (this->_stream.compare("__heartbeat") == 0)
                return true;
            return false;
        }
  

• shellbolt 调用 shell 命令时遇到 no such file 的异常

  • 原因1.：二进制和脚本放在src/main/resources/ 打包后不会上传到集群中，解决方案为放到src/main/resources/resources目录下
  • 原因2：使用 super("sh start.sh"); 整个字符串是作为一个文件路径进行查找的，实际应为 super("sh", "start.sh");

示例项目

https://gitee.com/wangpeibin/storm-cpp-integration