serialization - ソケットを使用して Flink で DataStream を送信します。連載問題

Question

VM からホスト マシンにデータのストリームを送信したいのですが、writeToSocket()以下に示す方法を使用しています。

joinedStreamEventDataStream.writeToSocket("192.168.1.10", 6998) ;

これjoinedStreamEventDataStreamは のタイプDataStream<Integer,Integer>です。

上記のメソッドにシリアライザを渡す方法を教えてください。

前もって感謝します

Answer 1

ソケットからデータを読み取る方法によって少し異なります。データの文字列表現であると予想される場合は、次の方法で実行できます。

joinedStreamEventDataStream.map(new MapFunction<Type, String>() {
    @Override
    public String map(Type value) throws Exception {
        return value.toString();
    }
}).writeToSocket(hostname, port, new SimpleStringSchema());

Flink のシリアル化形式を維持したい場合は、次のように記述できます。

joinedStreamEventDataStream.writeToSocket(
    hostname, 
    port, 
    new TypeInformationSerializationSchema<>(
        joinedStreamEventDataStream.getType(), 
        env.getConfig()));

独自のシリアル化形式で出力したい場合は、SerializationSchemaAlex が指摘したように独自のものを実装する必要があります。