Spring Cloud Dataflow（rabbit | processor | hdfs）出力バイナリ

Question

現在、RabbitMQメッセージバスからメッセージを取り出し、メッセージサイズ（バイト単位）を追加し、HDFSシンクを使用してメッセージを出力する必要があります。

開始するには、メッセージにサイズを付加する独自のプロセッサを作成しました。私がこれをする理由は、エンコーディングがGoogleのプロトコルバッファである必要があるためです。

私のアプリは、次のようになります。

stream create --name rabbit-to-hdfs --definition "rabbit | delim-protobuf | hdfs " 

HDFSシンクは、私は12768762 @ [Bを参照してくださいメッセージを出力します。私の周りGoogle'dしており、以下を追加するための推奨事項を見てきました：

spring.cloud.stream.bindings.input.consumer.headerMode=raw 

しかし、これがすべてで私を助けるために見ていません！つまり、次のようにファイルを表示するようにアプリを変更した場合：

[input | processor ] | file --binary=true 

これで問題なく動作します。しかし、私はHDFSシンクで提供されるロールオーバー機能が気に入っています。

アイデア？

Answer 1

ファイルは、受信したバイトをダンプするだけなので、HDFSシンクを見ると、入力としてjava.io.Serializableオブジェクトを消費する必要があるようです。しかし、あなたの場合は、protobufオブジェクトからバイト配列を送信しています（私はそれが起こっていると仮定しています）

Answer 2

型は互換性がありません。そのcontentTypeをSCSに設定することによって、java直列化を使用してwriteObjectを呼び出すようにフレームワークに要求するだけです。しかし、すでにシリアル化フレームワークであるprotobufを使用しているので、動作しません。ここで問題となるのは、シンクはシンクロコードを期待する（シンクコードに慣れていない）ようですが、シンクロナイズはできません。あなたができることは、シンクアプリを修正するか、またはprotobufからSerializableに変換する方法を知っているカスタムコンバータを提供することです。それが正直であると分かるかどうかは分かりません。