多语言协议（如何为其它语言提供支持）

优质

小牛编辑

129浏览

2023-12-01

本页介绍了Storm 0.7.1中的多语言协议。0.7.1之前的版本使用了一个有些不同的协议，文档位于 [here](Storm-multi-language-protocol-(versions-0.7.0-and-below).html).

Storm 多语言协议

Shell 组件

通过ShellBolt,ShellSpout和ShellProcess类实现对多语言的支持。这些类实现IBolt和ISpout接口以及执行脚本的协议或程序通过Shell使用Java的ProcessBuilder类。

包装Shell脚本

默认情况下，ShellPorcess假定您的代码打包在您的jar的resources子目录下的拓扑Jar内，默认情况会更改当前的工作目录，该可执行线程是从Jar中提取的资源目录。一个Jar没有存储其中文件的权限。这包括允许Shell脚本由操作系统加载和运行的执行位。因此，在大多数示例中，脚本具有python mybolt.py的形式，因为python可执行文件已经在主管上，mybolt忆打包在jar的资源目录中。

如果你想打包更复杂的东西，像一个新版本的python本身，你需要改用blod这个存储和一个支持权限的.tgz 档案。

可以看这个文档 Blob Store 有更加详细的说明怎么运送jar的细节。

使用ShellBolt/ShellSpout与可执行文件+脚本一起发布在blod store cache中。

changeChildCWD(false);

在ShellBolt/ShellSpout的构造函数中。shell命令将相对于工作者的cwd。哪里的资源链接。

所以如果我发送python与一个名为newPython和一个python ShellSpout的符号链接我并发送到shell_spout.py，我会有如下写法

public MyShellSpout() {
    super("./newPython/bin/python", "./shell_spout.py");
    changeChildCWD(false);
}

输出字段

输出字段是Thrift拓扑定义的一部分。这就意味着当您在java中的multing时，您需要创建一个扩展ShellBolt的bolt,实现IRichBolt,并声明declareOutputFields(类似于ShellSpout)中的字段。

您可以学习更多关于 Concepts

协议序言

一个简单的协议是通过STDIN和STDOUT来实现的执行脚本或程序。与该过程交换的所有数据为JSON格式，几乎可以支持任何语言。

包装你的东西

要在集群上运行Shell组件，那就是shelled的脚本必须在jar中提供的resources/目录中给master。

但是，在本地机器的开发或测试过程中，资源目录只需要在类路径中。

协议

Notes:

该协议的两端使用线读机制，所以一定要从输入中剪掉换行符并将其追加到输出中。
所有JSON输入和输出都由包含"end"的单行终止。请注意，此分隔符本身不是JSON编码的。
下面的项目符号是从脚本作者的角度编写的STDIN和STDOUT。

初始化握手

两种类型的shell组件的初始化握手是相同的：

STDIN: 设置信息。这是一个具有Storm配置，PID目录和拓扑上下文的JSON对象，像这样：

{
    "conf": {
        "topology.message.timeout.secs": 3,
        // etc
    },
    "pidDir": "...",
    "context": {
        "task->component": {
            "1": "example-spout",
            "2": "__acker",
            "3": "example-bolt1",
            "4": "example-bolt2"
        },
        "taskid": 3,
        // Everything below this line is only available in Storm 0.10.0+
        "componentid": "example-bolt"
        "stream->target->grouping": {
        	"default": {
        		"example-bolt2": {
        			"type": "SHUFFLE"}}},
        "streams": ["default"],
 		"stream->outputfields": {"default": ["word"]},
	    "source->stream->grouping": {
	    	"example-spout": {
	    		"default": {
	    			"type": "FIELDS",
	    			"fields": ["word"]
	    		}
	    	}
	    }
	    "source->stream->fields": {
	    	"example-spout": {
	    		"default": ["word"]
	    	}
	    }
	}
}

您的脚本应该在此目录中创建一个以其PID命名的空文件。例如，PID为1234，因此在目录中创建名为1234的空文件。这个文件让主管知道PID,以便稍后关闭该过程。

从Storm 0.10.0起，Storm发送到shell组件的上下文一直是大大增强包括可用于JVM组件的拓扑上下文的所有方面。一个关键的补充是能够确定拓扑结构中的shell组件的源和目标（即输入和输出）stream->target->grouping and source->stream->grouping 字典。在这些嵌套字典的最内层，分组被表示为一个最低限度具有type键的字典，但也可以有一个fields键，指定FIELDS分组中涉及哪些字段。

STDOUT: 你的PID，在JSON对象中，像 {"pid": 1234}。shell组件将PID记录到其日志中。

接下来会发生什么取决于组件的类型：

Spouts

Shell spouts 是同步的. 其余的发生在一段时间(true)循环：

STDIN: 下一个，ack,激活，停用或失败命令。

"next" 相当于ISpout's的nextTuple。看起来就像：

{"command": "next"}

"ack" 看起来像:

{"command": "ack", "id": "1231231"}

"activate" 相当于ISpout's的 activate:

{"command": "activate"}

"deactivate" 相当于ISpout's的 deactivate:

{"command": "deactivate"}

"fail" 看起来像:

{"command": "fail", "id": "1231231"}

STDOUT: 您以前命令的输出结果。这可以是一系列发射和日志。

An emit looks like:

{
	"command": "emit",
	// The id for the tuple. Leave this out for an unreliable emit. The id can
    // be a string or a number.
	"id": "1231231",
	// The id of the stream this tuple was emitted to. Leave this empty to emit to default stream.
	"stream": "1",
	// If doing an emit direct, indicate the task to send the tuple to
	"task": 9,
	// All the values in this tuple
	"tuple": ["field1", 2, 3]
}

如果不直接执行emit,则将立即收到STDIN上以元数组发布的元组为JSON数组。

"log" 将在工作日志中记录一条消息。看起来像：

{
	"command": "log",
	// the message to log
	"msg": "hello world!"
}

STDOUT: "sync"命令结束发射和日志的顺序。看起来像：

{"command": "sync"}

同步之后，ShellSpout将不会读取您的输出，直到它发送另一个next,ack，或fail命令。

请注意，与ISpout类似，工作人员的所有spouts将在下一次，确认或失败后被锁定，直到您同步。也像ISpout,如果没有元组为下一个发出，您应该睡眠少量的时间才能同步。ShellSpout不会自动为您做睡眠。

Bolts

The shell bolt 是异步的. 您将在STDIN上收到元组，只要它们可用，您可以发出，确认或失败，并随时通过写入SDTOUT，如下所示:

STDIN: 一个元组！这是一个这样的JSON编码结构:

{
    // The tuple's id - this is a string to support languages lacking 64-bit precision
	"id": "-6955786537413359385",
	// The id of the component that created this tuple
	"comp": "1",
	// The id of the stream this tuple was emitted to
	"stream": "1",
	// The id of the task that created this tuple
	"task": 9,
	// All the values in this tuple
	"tuple": ["snow white and the seven dwarfs", "field2", 3]
}

STDOUT: An ack, fail, emit, or log. Emits look like:

{
	"command": "emit",
	// The ids of the tuples this output tuples should be anchored to
	"anchors": ["1231231", "-234234234"],
	// The id of the stream this tuple was emitted to. Leave this empty to emit to default stream.
	"stream": "1",
	// If doing an emit direct, indicate the task to send the tuple to
	"task": 9,
	// All the values in this tuple
	"tuple": ["field1", 2, 3]
}

如果不直接执行emit，那么您将会收到在STDIN上发布元组的任务ids作为JSON数组。请注意，由于异步性质的shell bolt协议，当你读后你可以收不到任务的ids。你可以改为阅读要处理的先前发布或新元组的任务ids。但是你将按照相应的排放顺序接收任务id列表。

An ack 看起来像:

{
	"command": "ack",
	// the id of the tuple to ack
	"id": "123123"
}

A fail 看起来像:

{
	"command": "fail",
	// the id of the tuple to fail
	"id": "123123"
}

A "log" 将在工作日志中记录一条消息。看起来像：

{
	"command": "log",
	// the message to log
	"msg": "hello world!"
}

请注意，从0.7.1版本起，不再需要一个shell bolt进行 '同步'操作。

心跳处理 (0.9.3 及以上)

直到Storm 0.9.3,心跳在ShellSpout/ShellBolt与它们之间多个子进程检测挂/子进程。任何通过多镜头与Storm进行连接的库，必须对听筒采取以下措施：

Spout

Shell spouts 是同步的,因此子流程总是在next()的末尾发送sync命令，所以你不必为支持spouts的心跳做很多工作。也就是说，在next()期间，不要让子进程睡眠超过工作超时。

Bolt

Shell bolts 是异步的, 因此ShellBolt将定期向其子进程发送心跳元组。心跳元组看起来像：

{
	"id": "-6955786537413359385",
	"comp": "1",
	"stream": "__heartbeat",
	// this shell bolt's system task id
	"task": -1,
	"tuple": []
}

当子进程接收到心跳元组时，它必须发送一个sync命令回到ShellBolt。