Spark Streaming 中的批量大小答案

【问题标题】：Batch Size in Spark StreamingSpark Streaming 中的批量大小
【发布时间】：2015-09-14 17:08:39
【问题描述】：

我是 Spark 和 Spark Streaming 的新手。我正在处理 Twitter 流数据。我的任务涉及独立处理每条推文，例如计算每条推文中的字数。根据我的阅读，Spark Streaming 中 RDD 上的每个输入批处理表单。因此，如果我给出 2 秒的批处理间隔，那么新的 RDD 包含两秒内的所有推文，并且应用的任何转换都将应用于整个两秒的数据，并且无法在这两秒内处理单个推文。我的理解正确吗？或者每条推文都会形成一个新的RDD？我有点迷茫……

【问题讨论】：

标签： scala twitter apache-spark twitter4j spark-streaming

【解决方案1】：

在一个批次中，您有一个 RDD，其中包含以 2 秒为间隔的所有状态。然后您可以单独处理这些状态。这是一个简单的例子：

 JavaDStream<Status> inputDStream = TwitterUtils.createStream(ctx, new OAuthAuthorization(builder.build()), filters);

      inputDStream.foreach(new Function2<JavaRDD<Status>,Time,Void>(){
            @Override
            public Void call(JavaRDD<Status> status, Time time) throws Exception {
                List<Status> statuses=status.collect();
                for(Status st:statuses){
                     System.out.println("STATUS:"+st.getText()+" user:"+st.getUser().getId());                      
                //Process and store status somewhere
                }
                return null;
            }});         
     ctx.start();
        ctx.awaitTermination();      
}

希望我没有误解你的问题。

卓然

【讨论】：

谢谢。如果我将状态单独存储在列表中，我是否可以在列表中应用所有 RDD 转换或操作，如 reduceByKey()、countByValue？虽然我是 Scala 新手，但我需要在 Scala 中完成。
我刚刚给你举了一个带有列表的例子，向你展示你可以访问单个状态，但是如果你想使用 spark 进一步处理它，你不应该收集状态来列出。例如，您可以实现 inputDStream.mapToPair 函数，该函数将通过某些键返回状态，例如用户 ID 或您需要的任何内容。然后你可以reduceByKey。不幸的是，我只有 Scala 的基本知识，不能给你举例，但是你可以用 Java 做的所有事情，你也可以用 Scala 做。
我想我可以将特定批次的状态存储在一个列表中，并使用 parallelize() 将该列表转换为 RDD，以便我可以应用 Spark 转换和操作。
如果你看前面的例子，你会注意到一批进来的这些状态已经是RDD，所以不需要像我在例子中那样将它转换为列表，然后并行化再次。如果您需要对一批状态进行 Spark 转换，则可以直接在函数调用方法中的 RDD 状态上进行。所以如果你想reduceByKey，你应该先PairRDD pairs=status.mapToPair...，然后pairs.reduceByKey...