Storm入門之第7章使用非JVM語言開發

阿新 • • 發佈：2018-12-22

有時候你可能想使用不是基於JVM的語言開發一個Storm工程，你可能更喜歡使用別的語言或者想使用用某種語言編寫的庫。

Storm是用Java實現的，你看到的所有這本書中的spout和bolt都是用java編寫的。那麼有可能使用像Python、Ruby、或者JavaScript這樣的語言編寫spout和bolt嗎？答案是當然

可以！可以使用多語言協議達到這一目的。

多語言協議是Storm實現的一種特殊的協議，它使用標準輸入輸出作為spout和bolt程序間的通訊通道。訊息以JSON格式或純文字格式在通道中傳遞。

我們看一個用非JVM語言開發spout和bolt的簡單例子。在這個例子中有一個spout

產生從1到10,000的數字，一個bolt過濾素數，二者都用PHP實現。

NOTE: 在這個例子中，我們使用一個很笨的辦法驗證素數。有更好當然也更復雜的方法，它們已經超出了這個例子的範圍。

有一個專門為Storm實現的PHP DSL(譯者注：領域特定語言)，我們將會在例子中展示我們的實現。首先定義拓撲。

...
TopologyBuilder builder = new TopologyBuilder();
builder.setSpout("numbers-generator", new NumberGeneratorSpout(1, 10000));
builder.setBolt("prime-numbers-filter", new
PrimeNumbersFilterBolt()).shuffleGrouping("numbers-generator");
StormTopology topology = builder.createTopology();
...

NOTE：有一種使用非JVM語言定義拓撲的方式。既然Storm拓撲是Thrift架構，而且Nimbus是一個Thrift守護程序，你就可以使用任何你想用的語言建立並提交拓撲。但是這已經超出了本書的範疇了。

這裡沒什麼新鮮了。我們看一下NumbersGeneratorSpout的實現。

public class NumberGeneratorSpout extends ShellSpout implements IRichSpout {
    public NumberGeneratorSpout(Integer from, Integer to) {
       super("php", "-f", "NumberGeneratorSpout.php", from.toString(), to.toString());
    }
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("number"));
    }
    public Map<String, Object> getComponentConfiguration() {
        return null;
    }
}

你可能已經注意到了，這個spout繼承了ShellSpout。這是個由Storm提供的特殊的類，用來幫助你執行並控制用其它語言編寫的spout。在這種情況下它告訴Storm如何執行你的PHP指令碼。

NumberGeneratorSpout的PHP指令碼向標準輸出分發元組，並從標準輸入讀取確認或失敗訊號。

在開始實現NumberGeneratorSpout.php指令碼之前，多觀察一下多語言協議是如何工作的。

spout按照傳遞給構造器的引數從from到to順序生成數字。

接下來看看PrimeNumbersFilterBolt。這個類實現了之前提到的殼。它告訴Storm如何執行你的PHP指令碼。Storm為這一目的提供了一個特殊的叫做ShellBolt的類，你惟一要做的事就是指出如何執行指令碼以及宣告要分發的屬性。

public class PrimeNumbersFilterBolt extends ShellBolt implements IRichBolt {
    public PrimeNumbersFilterBolt() {
        super("php", "-f", "PrimeNumbersFilterBolt.php");
    }
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("number"));
    }
}

在這個構造器中只是告訴Storm如何執行PHP指令碼。它與下列命令等價。

    php -f PrimeNumbersFilterBolt.php

PrimeNumbersFilterBolt.php指令碼從標準輸入讀取元組，處理它們，然後向標準輸出分發、確認或失敗。在開始這個指令碼之前，我們先多瞭解一些多語言協議的工作方式。

發起一次握手
開始迴圈
讀/寫元組

NOTE：有一種特殊的方式可以使用Storm的內建日誌機制在你的指令碼中記錄日誌，所以你不需要自己實現日誌系統。

下面我們來看一看上述每一步的細節，以及如何用PHP實現它。

發起握手

為了控制整個流程（開始以及結束它），Storm需要知道它執行的指令碼程序號（PID）。根據多語言協議，你的程序開始時發生的第一件事就是Storm要向標準輸入（譯者注：根據上下文理解，本章提到的標準輸入輸出都是從非JVM語言的角度理解的，這裡提到的標準輸入也就是PHP的標準輸入）傳送一段JSON資料，它包含Storm配置、拓撲上下文和一個程序號目錄。它看起來就像下面的樣子：

{
    "conf": {
        "topology.message.timeout.secs": 3,
        // etc
    },
    "context": {
        "task->component": {
            "1": "example-spout",
            "2": "__acker",
            "3": "example-bolt"
        },
        "taskid": 3
    },
    "pidDir": "..."
}

指令碼程序必須在pidDir指定的目錄下以自己的程序號為名字建立一個檔案，並以JSON格式把程序號寫到標準輸出。

{"pid": 1234}

舉個例子，如果你收到/tmp/example\n而你的指令碼程序號是123，你應該建立一個名為/tmp/example/123的空檔案並向標準輸出列印文字行 {“pid”: 123}\n（譯者注：此處原文只有一個n，譯者猜測應是排版錯誤）和end\n。這樣Storm就能持續追蹤程序號並在它關閉時殺死指令碼程序。下面是PHP實現：

$config = json_decode(read_msg(), true);
$heartbeatdir = $config['pidDir'];
$pid = getmypid();
fclose(fopen("$heartbeatdir/$pid", "w"));
storm_send(["pid"=>$pid]);
flush();

你已經實現了一個叫做read_msg的函式，用來處理從標準輸入讀取的訊息。按照多語言協議的宣告，訊息可以是單行或多行JSON文字。一條訊息以end\n結束。

function read_msg() {
    $msg = "";
    while(true) {
        $l = fgets(STDIN);
        $line = substr($l,0,-1);
        if($line=="end") {
            break;
        }
        $msg = "$msg$line\n";
    }
    return substr($msg, 0, -1);
}
function storm_send($json) {
    write_line(json_encode($json));
    write_line("end");
}
function write_line($line) {
    echo("$line\n");
}

NOTE：flush()方法非常重要；有可能字元緩衝只有在積累到一定程度時才會清空。這意味著你的指令碼可能會為了等待一個來自Storm的輸入而永遠掛起，而Storm卻在等待來自你的指令碼的輸出。因此當你的指令碼有內容輸出時立即清空緩衝是很重要的。

開始迴圈以及讀/寫元組

這是整個工作中最重要的一步。這一步的實現取決於你開發的spout和bolt。

如果是spout，你應當開始分發元組。如果是bolt，就迴圈讀取元組，處理它們，分發它發，確認成功或失敗。

下面我們就看看用來分發數字的spout。

$from = intval($argv[1]);
$to = intval($argv[2]);
while(true) {
    $msg = read_msg();
    $cmd = json_decode($msg, true);
    if ($cmd['command']=='next') {
        if ($from<$to) {
            storm_emit(array("$from"));
            $task_ids = read_msg();
            $from++;
        } else {
            sleep(1);
        }
    }
    storm_sync();
}

從命令列獲取引數from和to，並開始迭代。每次從Storm得到一條next訊息，這意味著你已準備好分發下一個元組。

一旦你傳送了所有的數字，而且沒有更多元組可發了，就休眠一段時間。

為了確保指令碼已準備好傳送下一個元組，Storm會在傳送下一條之前等待sync\n文字行。呼叫read_msg()，讀取一條命令，解析JSON。

對於bolts來說，有少許不同。

while(true) {
    $msg = read_msg();
    $tuple = json_decode($msg, true, 512, JSON_BIGINT_AS_STRING);
    if (!empty($tuple["id"])) {
        if (isPrime($tuple["tuple"][0])) {
            storm_emit(array($tuple["tuple"][0]));
        }
        storm_ack($tuple["id"]);
    }
}

迴圈的從標準輸入讀取元組。解析讀取每一條JSON訊息，判斷它是不是一個元組，如果是，再檢查它是不是一個素數，如果是素數再次分發一個元組，否則就忽略掉，最後不論如何都要確認成功。

NOTE：在json_decode函式中使用的JSON_BIGINT_AS_STRING是為了解決一個在JAVA和PHP之間的資料轉換問題。JAVA傳送的一些很大的數字，在PHP中會丟失精度，這樣就會導致問題。為了避開這個問題，告訴PHP把大數字當作字串處理，並在JSON訊息中輸出數字時不使用雙引號。PHP5.4.0或更高版本要求使用這個引數。

emit，ack，fail，以及log訊息都是如下結構：

emit

{
    "command": "emit",
    "tuple": ["foo", "bar"]
}

其中的陣列包含了你分發的元組資料。

ack

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

其中的id就是你處理的元組的ID。
fail

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

與ack（譯者注：原文是emit從上下JSON的內容和每個方法的功能上判斷此處就是ack，可能是排版錯誤）相同，其中id就是你處理的元組ID。
log

{
    "command": "log",
    "msg": "some message to be logged by storm."
}

下面是完整的的PHP程式碼。

//你的spout：
<?php
function read_msg() {
    $msg = "";
    while(true) {
        $l = fgets(STDIN);
        $line = substr($l,0,-1);
        if ($line=="end") {
            break;
        }
        $msg = "$msg$line\n";
    }
    return substr($msg, 0, -1);
}
function write_line($line) {
    echo("$line\n");
}
function storm_emit($tuple) {
    $msg = array("command" => "emit", "tuple" => $tuple);
    storm_send($msg);
}
function storm_send($json) {
    write_line(json_encode($json));
    write_line("end");
}
function storm_sync() {
    storm_send(array("command" => "sync"));
}
function storm_log($msg) {
    $msg = array("command" => "log", "msg" => $msg);
    storm_send($msg);
    flush();
}
$config = json_decode(read_msg(), true);
$heartbeatdir = $config['pidDir'];
$pid = getmypid();
fclose(fopen("$heartbeatdir/$pid", "w"));
storm_send(["pid"=>$pid]);
flush();
$from = intval($argv[1]);
$to = intval($argv[2]);
while(true) {
    $msg = read_msg();
    $cmd = json_decode($msg, true);
    if ($cmd['command']=='next') {
        if ($from<$to) {
            storm_emit(array("$from"));
            $task_ids = read_msg();
            $from++;
        } else {
            sleep(1);
        }
    }
    storm_sync();
}
?>
//你的bolt:
<?php
function isPrime($number) {
    if ($number < 2) {
        return false;
    }
    if ($number==2) {
        return true;
    }
    for ($i=2; $i<=$number-1; $i++) {
        if ($number % $i == 0) {
            return false;
        }
    }
    return true;
}
function read_msg() {
    $msg = "";
    while(true) {
        $l = fgets(STDIN);
        $line = substr($l,0,-1);
        if ($line=="end") {
            break;
        }
        $msg = "$msg$line\n";
    }
    return substr($msg, 0, -1);
}
function write_line($line) {
    echo("$line\n");
}
function storm_emit($tuple) {
    $msg = array("command" => "emit", "tuple" => $tuple);
    storm_send($msg);
}
function storm_send($json) {
    write_line(json_encode($json));
    write_line("end");
}
function storm_ack($id) {
    storm_send(["command"=>"ack", "id"=>"$id"]);
}
function storm_log($msg) {
    $msg = array("command" => "log", "msg" => "$msg");
    storm_send($msg);
}
$config = json_decode(read_msg(), true);
$heartbeatdir = $config['pidDir'];
$pid = getmypid();
fclose(fopen("$heartbeatdir/$pid", "w"));
storm_send(["pid"=>$pid]);
flush();
while(true) {
    $msg = read_msg();
    $tuple = json_decode($msg, true, 512, JSON_BIGINT_AS_STRING);
    if (!empty($tuple["id"])) {
        if (isPrime($tuple["tuple"][0])) {
            storm_emit(array($tuple["tuple"][0]));
        }
        storm_ack($tuple["id"]);
    }
}
?>

NOTE：需要重點指出的是，應當把所有的指令碼檔案儲存在你的工程目錄下的一個名為multilang/resources的子目錄中。這個子目錄被包含在傳送給工人程序的jar檔案中。如果你不把指令碼包含在這個目錄中，Storm就不能執行它們，並丟擲一個錯誤。

Storm入門之第7章使用非JVM語言開發

Storm入門之第7章使用非JVM語言開發

Storm入門之第五章Bolts

Storm入門之第四章Spouts

Storm入門之第三章拓撲

Storm入門之第6章一個實際的例子

Storm入門之第8章事務性拓撲

《web安全之機器學習入門》第7章樸素貝葉斯模型檢測webshell

《Java從小白到大牛》之第7章控制語句

數據結構學習之第7章樹和二叉樹

C語言程式設計入門之--第三章編寫第一個C語言程式

C語言程式設計入門之--第四章C語言基本資料型別

C語言程式設計入門之--第五章C語言基本運算和表示式-part1

C語言程式設計入門之--第五章C語言基本運算和表示式-part2

C語言程式設計入門之--第五章C語言基本運算和表示式-part3

python編程快速上手之第15章實踐項目參考答案(17.7.2)

python編程快速上手之第15章實踐項目參考答案(17.7.3)

算法入門經典-第五章例題5-7 醜數

算法入門經典-第七章例題7-1 除法

算法入門經典-第七章例題7-2-2 可重集的排列

Unity Shader入門精要學習筆記 - 第14章非真實感渲染

Storm入門之第7章使用非JVM語言開發

相關推薦