# 多語言協議 本頁介紹了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](distcache-blobstore.html) 有更加詳細的說明怎么運送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](Concepts.html) ## 協議序言 一個簡單的協議是通過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。