kafkaストリーミング最初のバッチ内のすべてのデータを読み取る

Question

カフカのトピックに50レコードを挿入しました。各レコードの間に1秒の待ち時間があり、50秒以上で50レコードになります。

このトピックのレコードを消費しているうちに、以下のコードでわかるように、バッチ間隔を1秒に保っています。理想的には〜50のRDDが必要です。 私はforeachRDDを使用してバッチでデータを取得し、各RDDのデータを処理しています。 以下のコードでは、 "call"メソッドが呼び出されるたびに、私は各RDDのレコードを数えています。驚くべきことは、最初のRDD自体が50個のレコードを表示しており、後続のRDDが0個のレコードを表示していることです。 私はこの動作を理解していません。理想的にはRDDあたり1レコードです。

私の理解が間違っていると誰も示唆できますか？

JavaStreamingContext jssc = new JavaStreamingContext（sparkConf、Durations.seconds（1））;

messages.foreachRDD(new VoidFunction<JavaRDD<ConsumerRecord<K, String>>>() { 
      public void call(JavaRDD<ConsumerRecord<K, V>> rdd) { 
       System.out.println("NUmber of elements in RDD : "+ rdd.count()); 

       List<Row> rows = rdd.map(record -> processData(record)) 
         .reduce((rows1, rows2) -> { 
          rows1.addAll(rows2); 
          return rows1; 
         }); 

       StructType schema = DataTypes.createStructType(fields); 
       Dataset ds = ss.createDataFrame(rows, schema); 
       ds.createOrReplaceTempView("trades");     
       ds.show(); 
      } 
     }); 

Answer 1

これはあなたが見ていることを引き起こすと予想される動作です。ストリーミングジョブを開始し、カフカのトピックにデータを入力した場合（1秒ごとに1レコード）、バッチサイズは期待どおりになっていました。 （未かなり...のでカフカのトピックパーティション）

しかし、あなたが、それはカフカのパーティションごとに動作することを心に留めておく必要がありmax.rate.per.partition

と呼ばれるスパークストリーミングパラメータを使用して、あなたが望むものを達成することが可能です。

例：

あなたのカフカのトピックに3つのパーティションがある場合。あなたのbatch size is 1 second。 max.rate.per.partition = 1

以下の構成では、RDDバッチごとに3つのアイテムが得られます。

サンプルストリーミングコンテキスト：

``` 
val sparkConf = new SparkConf(). 
    set("spark.streaming.kafka.maxRatePerPartition", 
     inputParam.maxRatePerPartition) 

val ssc = new StreamingContext(sparkConf, inputParam.batchDuration) 
``` 

あなたのカフカのトピックは1つのパーティションがある場合、あなたはあなたが望むものを正確に取得します。

Answer 2

スパークストリーミングbatch intervalデータ到着又は収集される間に、ある「マイクロバッチ」を使用します。

各バッチ間隔の締め切り時に、その時点までに到着したデータが処理のためにSparkに送信されます。ストリーミングプロセスが開始されると、すべてのデータがすでにKafkaで受信されていれば、すべてのデータが最初の間隔で一度に処理されます。その瞬間の後に到着

最終的に新しいデータがさらに間隔でのプロセスになります。