Spark Streaming/Spark 是否像 main() 方法中的 while 循环一样工作？

Question

这是一个非常直接的问题：

我正在通过以下方式使用 Spark Streaming：

private static int count=0;

public static void main(String[] args) throws Exception {
    if (args.length < 2) {
      System.err.println("Usage: sparkstreaminggetjson <hostname> <port>");
      System.exit(1);
    }

 SparkConf sparkConf = new  
SparkConf().setMaster("local[2]").setAppName("sparkstreaminggetjson");

JavaSparkContext sc=new JavaSparkContext(sparkConf);

JavaStreamingContext ssc = new JavaStreamingContext(sparkConf, 
Durations.seconds(1));

JavaReceiverInputDStream<String> lines = ssc.socketTextStream(

args[0],Integer.parseInt(args[1]), StorageLevels.MEMORY_AND_DISK_SER);

count=count+1;

lines.print;

ssc.start();

ssc.awaitTermination();
}

每次新的“批处理时间”开始时，变量行从套接字获取新值并打印出来，变量计数增加 1。

我的问题是：现在我在 main() 之外声明计数，因为如果我这样做

public static void main(String[] args) throws Exception {
    if (args.length < 2) {
      System.err.println("Usage: sparkstreaminggetjson <hostname> <port>");
      System.exit(1);
    }

 //declare count here
 int count=0;
 SparkConf sparkConf = new  
SparkConf().setMaster("local[2]").setAppName("sparkstreaminggetjson");

JavaSparkContext sc=new JavaSparkContext(sparkConf);

JavaStreamingContext ssc = new JavaStreamingContext(sparkConf, 
Durations.seconds(1));

JavaReceiverInputDStream<String> lines = ssc.socketTextStream(

args[0],Integer.parseInt(args[1]), StorageLevels.MEMORY_AND_DISK_SER);

count=count+1;

lines.print;

ssc.start();

ssc.awaitTermination();
}

每次新的批处理时间开始时，变量计数都会重置为 0。所以，它基本上就像一个 while 循环。

我有什么方法可以在 main() 方法中声明 count 以使 Spark Streaming 循环不会将其重置为 0？如果是这样，我在哪里声明它？

基本上我想知道当我们创建一个 Spark 上下文时，我们是否让整个 main() 方法表现得像一个 while 循环，或者这个循环是否有一个特定的地方开始。

非常感谢，希望我没有混淆。

Answer 1

没有。 Spark Streaming 不会在“main”上执行“while 循环”。 Spark Streaming 使用调度程序以提供的批处理间隔触发注册的output operators。

这些输出运算符将在每个批处理间隔触发底层 RDD 的实现。在这个物化过程中，RDD操作将在Spark集群中执行。

使用与集群中序列化的任何代码交互的全局静态变量将导致意外行为。

Spark 是一个分布式计算框架，它的操作就是以这个目标为导向的。 “map”、“filter”等经典转换将在集群中分布在节点上的数据片段上执行。

Spark 中与“全局变量”最接近的等价物是 broadcast variables，但它们不能在闭包中更新。

根据提供的代码，我了解到这些是了解 Spark Streaming 模型的初步实验。花一些时间研究related material 以充分了解它的工作原理。在这种特殊情况下，增加静态变量只能在本地模式下工作，因为所有代码都在同一个 JVM 上执行，但这不是 Spark Streaming 的目标。