問題タブ [dstream]

たとえば、どのように

とは異なる

ここにあることに注意しxてDStreamください。

以下のスニペットでは、(Kafka から受け取った) 温度の DStream を pandas Dataframe に変換しようとしています。

そのままでは、平均が計算されることはありません。これは、「df」が pandas データフレーム (?) ではないためだと思います。

df = spark.createDataFrame(df.toPandas())関連ドキュメントに従って使用してみましたが、コンパイラは「toPandas()」を認識せず、変換は行われません。

私は正しい道を進んでいますか? もしそうなら、どのように変換を適用すればよいですか?

それとも、私のアプローチが間違っているので、DStream を別の方法で処理する必要がありますか?

前もって感謝します！

文字列のリストがありますが、リストをスパーク ストリーミングの DStream に変更する方法が見つかりません。私はこれを試しました：

しかし、日食はsparkContextがsqlContextのメンバーではないと言っているので、どうすればこれを行うことができますか? よろしくお願いします。

ストリームが時間 X に受信されたとします。バッチ期間が 1 分であるとします。現在、エグゼキュータは最初のバッチを処理しています。ただし、この実行には X+3 まで 3 分かかります。しかし、X+1 と X+2 では、他の 2 つのバッチを受け取ります。X+1 で最初のバッチが失われるということですか? それとも私の記憶に保存されていて、まだ処理中ですか？

DStream[String, Int単語数のペアを持つ] があり("hello" -> 10)ます。これらのカウントをステップ インデックスを使用して cassandra に書き込みたいと考えています。インデックスは次のように初期化されvar step = 1、マイクロバッチが処理されるたびに増分されます。

次のように作成された cassandra テーブル:

ストリームをテーブルに書き込もうとすると...

...わかりjava.lang.IllegalArgumentException: Some primary key columns are missing in RDD or have not been selected: stepました。

step3 つの列をまとめて書き込むために、ストリームの先頭にインデックスを追加するにはどうすればよいですか?

私はspark 2.0.0、scala 2.11.8、cassandra 3.4.0、およびspark-cassandra-connector 2.0.0-M3を使用しています。

私はかなり新しい Spark Streaming

2 つの値 x y を含むストリーミング データがあります。例えば

1 300

2 8754

3 287

等

ストリーミングされたデータから、最小の y 値、最大の y 値、および x 値の平均を取得したいと考えています。これは次のように出力する必要があります (上記の例を使用)。

287 8754 4

個々の変換/縮小でこれらの値を計算できましたが、単一の変換ではできませんでした

これが私の現在のコードです

私はこのstackoverflowの質問を経験しました。回答によると、バッチ間隔にDStream1つだけ作成されます。RDD

例えば：

私のバッチ間隔は 1 分で、Spark ストリーミング ジョブは Kafka トピックからのデータを消費しています。

私の質問は、DStream で利用可能な RDD は、最後の 1 分間のデータ全体をプル/含むか? 過去 1 分間に作成されたすべてのデータを取得するために設定する必要がある条件やオプションはありますか?

3 つのパーティションを持つ Kafka トピックがあり、3 つのパーティションすべてに最後の 1 分間のデータが含まれている場合、DStream はすべての Kafka トピック パーティションで最後の 1 分間に作成されたすべてのデータをプル/格納しますか?

アップデート：

DStream に複数の RDD が含まれているのはどの場合ですか?