structured-streaming

1熱

2答えて

データセット全体をロードする前に、各イベントにカスタムデータの書式設定/マップを適用する方法はありますか？

Apacheのスパークへのデータのストリームを読み込むためのdocumentationから標準的な方法は次のとおりです。 events = spark.readStream \ .format("json") \ # or parquet, kafka, orc... .option() \ # format specific options .schem

1熱

1答えて

Kafkaトピックでレコードが利用できない場合、StreamingQueryStatus.isDataAvailableがオンになるのはなぜですか？

私はspark構造化ストリーミングクエリを停止する必要があります。まずは、カフカで処理できるデータがあるかどうかチェックします。はいの場合は処理してからクエリを停止します。 if (query.status.isDataAvailable) { query.processAllAvailable() } query.stop() 私はJacek Lasko

2熱

2答えて

mapGroupsWithStateでstateSnapshotにアクセスする方法、またはストリーム間でGroupStateを共有する方法はありますか？

DStream APIでは、MapWithStateDStream.stateSnapshots()を使用してステートフルストリームのスナップショット状態にアクセスできます。新しいStructured Streaming APIでは、mapGroupsWithStateに渡された関数だけが状態にアクセスして更新できるようです。私は入力イベントに基づいてストリームのインメモリ分散状態を作成したいと

1熱

3答えて

カフカストラクチャードストリーミングチェックポイント

私はカフカから構造化ストリーミングをしようとしています。 HDFSにチェックポイントを格納する予定です。私はSparkストリーミングのためにHDFSにチェックポイントを保存しないことを推奨するclouderaブログを読んでいます。構造体ストリーミングチェックポイントでも同じ問題です。 https://blog.cloudera.com/blog/2017/06/offset-management-

1熱

2答えて

少なくともN個の行がある場合にのみストリーム処理を実行する方法は？

私はkafkaコンシューマで次のようなspark SQL/Streamingクエリを持っています。バッチサイズが特定のサイズNに達したときにフェッチを条件付きにするように指定するにはどうすればいいですか？そうしないと、 Dataset<VideoEventData> ds = spark .readStream() .format("kafka") .optio

0熱

1答えて

Spark Structured StreamingエグゼキュータとKafkaパーティション間のマッピングの説明

4つのパーティションを持つKafkaトピックで4人の作業者を含む構造化ストリームを配置しました。私は、作業員< - >パーティションの間に1対1のマッピングで4つのパーティションに4つの作業者が配置されていることを前提としていました。しかし、そうではありません。すべてのパーティションが同じExecutorによって提供されています。私はスレッドIDをチェックし、エグゼキュータをログすることでこれ

2熱

1答えて

ストリーミングデータセットをピボットする方法は？

Sparkストリーミングデータセット（構造化ストリーミング）をピボットしようとしていますが、AnalysisException（以下の抜粋）があります。誰かが、構造化されたストリーム（スパーク2.0）では実際にはピボットがサポートされていないことを確認できますか？スレッド「メイン」org.apache.spark.sql.AnalysisExceptionで例外：ストリーミングソースとクエリは

1熱

1答えて

Spark Structured Streaming用のTCPソケットを使用してデータを送信する方法

TCPソケットを使用してデータを作成して送信するので、Spark Structured Streamingのストリーミングソースとして使用できます。私はWindows OSを使用しています。私は自分のlocalhostにSSHのためにパテを使ってみましたが、それは私が何を与えるべきかわからないログインとパスワードを求めています。これを達成するために誰も助けてください。ありがとうございました！

4熱

1答えて

スパーク

私はそうのような構造化されたストリームを実装行かなければならなかった ... myDataSet .map(r => StatementWrapper.Transform(r)) .writeStream .foreach(MyWrapper.myWriter) .start() .awaitTermination() このすべてが動作している

1熱

1答えて

flatMapGroupsWithStateを使用してステートフルアグリゲーションを行う方法は？

flatMapGroupsWithStateの適用中に次のエラーメッセージが表示されます。スレッド「メイン」org.apache.spark.sql.AnalysisExceptionで例外：更新モードでflatMapGroupsWithStateは、ストリーミングデータフレーム/データセットの集計ではサポートされていません。以下は私がやろうとしていることです。カフカから読むメッセージ& そ