为什么 Spark Streaming 中 2 核和 4 核以及不同分区的处理时间相同？

Question

我正在尝试在我的 4 核机器上以本地模式运行有关 Spark 流应用程序处理时间的一些测试。

这是我的代码：

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

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


    JavaReceiverInputDStream<String> streamData1 = ssc.socketTextStream(args[0], Integer.parseInt(args[1]),
            StorageLevels.MEMORY_AND_DISK_SER);

    streamData1.print();

我每秒收到 1 条 JSON 消息。 所以，我测试了 4 种不同的场景：

1) setMaster(...local[2]) 和 1 个分区

2) setMaster(...local[*]) 和 1 个分区

3)setMaster(...local[2])和4个分区（使用streamData1.repartition(4)）

4) setMaster(...local[*]) 和 4 个分区（使用 streamData1.repartition(4)）

当我在 UI 中查看平均处理时间时，我得到的每个场景的平均处理时间如下：

1) 30 毫秒

2) 28 毫秒

3) 72 毫秒

4) 75 毫秒

我的问题是：为什么 1 和 2 以及 3 和 4 的处理时间几乎相同？ 我意识到例如从 2 增加到 4 是正常的，因为重新分区是一个随机操作。我不明白的是，例如在 4）中，为什么处理与 3 如此相似？由于我正在提高并行化级别，并且我有更多的内核可以将任务分配到，所以它不应该小得多吗？

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

Answer 1

其中一些取决于您的 JSON 消息的外观，我假设每条消息都是一个没有换行符的字符串。在这种情况下，每秒 1 条消息和 1 秒的批处理间隔，在每个批处理中，您将获得一个只有一个项目的 RDD。您不能将其拆分为多个分区，因此当您重新分区时，您仍然会在数据方面遇到相同的情况，但会产生重新分区步骤的开销。

即使有大量数据，当您对数据所做的所有操作都是 print() 时，我也不会期望有太大差异：这将占用您数据的前 10 项，如果它们只能来自一个分区，我希望 Spark 优化它以仅计算一个分区。无论如何，如果您显着增加每批次的数据量，并对整个数据集进行一些实际处理，您将获得更具代表性的数字，至少像 streamData1.count().print() 这样的。

为了更好地理解发生了什么，深入了解 Spark 用户界面的其他部分也很有用，例如 Stages 选项卡，它可以告诉您有多少执行时间是混洗、序列化等，而不是实际执行时间，以及影响性能的事物，例如告诉您哪些位可能被缓存的 DAG，以及 Spark 能够跳过的任务。