問題タブ [spring-cloud-dataflow]

Spring Cloud Streams と Dataflow を学び始めたばかりで、重要なユースケースの 1 つを知りたいです。メッセージを受け取り、それを 5 回再送信して出力するサンプル プロセッサ Multiplier を作成しました。

5 番目の送信前に、このプロセッサがクラッシュすることがわかります。なんで？それができるからです（プログラムは例外をスローします）。この場合、Spring Cloud Stream で障害防止を実践したかったのです。

私が達成したいのは、入力メッセージを DLQ に戻し、前に送信された 4 つのメッセージを元に戻し、次のオペランドで消費されないようにすることです (通常の JMS トランザクションと同様)。プロセッサ プロジェクトで次のプロパティを定義しようとしましたが、成功しませんでした。

可能かどうか、また何が間違っているのか教えていただけますか? いくつかの例を挙げていただければ幸いです。

Spring Cloud Data Flow - Cloudfoundry サーバー (v1.0.0.M4)

プロパティ ファイルで構成情報を外部化してから、Spring Cloud Config Server を使用してインストール時にこれらの環境設定を提供しようとしているときに、YML に通常は特定の方法で配置する値のいくつかについて質問があります。マニフェスト。

まず、YML マニフェストでは、次のように定義します。

では、これらをプロパティ ファイルに入れるにはどうすればよいでしょうか。これは私の推測です：

まだこのタスクに取り組もうとしています。私は次のものを持っています：

• Pivotal Cloud Foundry - エラスティック ランタイム 1.7.5
• Spring クラウド データ フロー サーバー Cloudfoundry - 1.0.0.M4
• Spring Cloud Config Server (SCCS) サービス (サービス名「scdf-sccs」)

SCCS インスタンスは BitBucket リポジトリに適切に接続されており、オンラインになっています。次のマニフェストで spring-cloud-dataflow-server-cloudfoundry-1.0.0.M4.jar をプッシュしようとしています。

私の BitBucket リポジトリには、環境変数を保持する次のプロパティ ファイルがあります: scdf-dev-abc.properties

プロパティ ファイルの内容は次のようになります。

マニフェストと SCCS バインディングを指定してアプリケーションをプッシュすると、プロパティ ファイルで指定されたプロパティが取得されず、問題のトラブルシューティングに苦労しています。私の構成、またはアプリケーション名とスプリングプロファイルを提供した方法に欠けているものはありますか?

構成がSpring Cloud Config Serverによって読み取られない理由を理解するために、他に何を探す必要がありますか? たとえば、Spring Cloud Data Flow アプリケーションが、BitBucket リポジトリにアクセスできないプロセスまたはユーザーとして実行されている場合、それは問題になりますか?

Spring Cloud Dataflow を使用してストリームを作成しようとしましたが、うまくいきませんでした (主にドキュメントがないため)。

問題 1: Docker 化された Spring Cloud Dataflow の Web GUI へのアクセス 基本的な Ubuntu コンテナーで Kafka を使用して実行されている Docker 化された Spring Cloud サーバーがあります。何らかの理由で、Windows の Web GUI (< docker-machine ip >:9393/dashboard) にアクセスできません。ただし、Nginx リバース プロキシを実行する別の Docker Ubuntu コンテナーがあり、< docker-machine ip >/index.html などに移動すると表示されます。ポートの問題ではないと思います。Spring -p 9393:9393 を使用したクラウド コンテナーのセットアップ。それ以外の場合、ポートは使用されません。

問題 2: JSON ヘッダーによるルーティング 私の最終的な目標は、Nginx から読み込まれたファイルを取得し、その JSON ヘッダー(2 つの異なる JSON ヘッダーがあります) に基づいてルーティングし、クエリを Cassandra に取り込むことです。JSONヘッダーによるソートを除いて、これをすべて行うことができます。どのアプリを使用することをお勧めしますか?

私は英語のネイティブ スピーカーではありませんが、質問をできるだけ明確に表現するように努めています。この問題に遭遇したため、2 日間混乱しましたが、まだ解決策が見つかりません。

Hadoop YARN の Spring Could Data Flow で実行されるストリームを構築しました。

ストリームは、HTTP ソース、プロセッサ、およびファイル シンクで構成されます。

1.Http ソース
HTTP ソース コンポーネントには、application.properties で定義された dest1 と dest2 という 2 つの異なる宛先にバインドする 2 つの出力チャネルがあります。

spring.cloud.stream.bindings.output.destination=dest1 spring.cloud.stream.bindings.output2.destination=dest2

以下は、参考用の HTTP ソースのコード スニペットです。

2. プロセッサ
プロセッサには、2 つの複数入力チャネルと、異なる宛先にバインドする 2 つの出力チャネルがあります。宛先バインディングは、プロセッサ コンポーネント プロジェクトの application.properties で定義されます。

以下は、プロセッサのコード スニペットです。

3. ファイル シンク コンポーネント。

Spring の公式の fil sink コンポーネントを使用します。maven://org.springframework.cloud.stream.app:file-sink-kafka:1.0.0.BUILD-SNAPSHOT

そして、applicaiton.properties ファイルに宛先バインディングを追加するだけです。spring.cloud.stream.bindings.input.destination=fileSink

4.発見：

私が期待したデータフローは次のようになります。

Source.handleRequest() -->Processor.handleRequest()

Source.handleRequest2() -->Processor.handleRequest2() --> Sink.fileWritingMessageHandler();

"processed by transform2" という文字列のみがファイルに保存されます。

しかし、私のテストの後、データフローは実際には次のようになります。

Source.handleRequest() -->Processor.handleRequest() --> Sink.fileWritingMessageHandler();

Source.handleRequest2() -->Processor.handleRequest2() --> Sink.fileWritingMessageHandler();

"processed by transform1" と "processed by transform2" の両方の文字列がファイルに保存されます。

5.質問:

Processor.handleRequest() の出力チャネルの宛先は、fileSink ではなく hdfsSink にバインドされますが、データは引き続きファイル Sink に流れます。私はこれを理解できず、これは私が望むものではありません。Processor.handleRequest2() からのデータのみが、両方ではなくファイル シンクに流れます。私が正しく行わない場合、誰かがそれを行う方法と解決策を教えてもらえますか? 2日間私を混乱させました。

ご親切にありがとうございました。

アレックス

Spring Cloud Data Flow で DLQ を構成しようとしています。これがストリームの定義とそれを展開する方法です

custom-transform - プロセッサ コードで、私は言及しました

つまり、メッセージに ERROR が含まれている場合、RunTimeException が発生し、これらのメッセージを DLQ でキャプチャしたいということです。しかし、コードを実行しているとき、test-tran という名前の Kafka DL キューを取得していないようです。

DLQ を有効にするにはさらにプロパティを設定する必要がありますか、それとも DLQ を適切に使用するためにコードを変更する必要がありますか?

カスタム変換コード

TransformationServiceApplication.java

TransformationMessageEndPoint.java

pom.xml

モジュールの追加

ストリームの追加

ストリームのデプロイ