Apache Spark join 操作的扩展能力较差

我在 Apache Spark 上运行“join”操作，发现没有弱可扩展性。如果有人能解释这一点，我将不胜感激。

我创建两个数据帧（“a”，“b”）和（“a”，“c”）并通过第一列连接数据帧。我为“一对一”连接生成数据帧值。另外，我使用相同的分区器来避免随机播放。

数据帧中的行数 - 1024 * 1024 * 16 * cores_total（cores_total - 启动程序的核心总数）。 “a”列由随机 Int 值组成，“b”列的所有值都等于 1，“c”列的所有值都等于 2。

理论上，随着数据大小和内核增加64倍，执行时间应该保持不变，但执行时间略有增长。我得到以下执行时间：

Apache Spark 版本 - 2.1.0。我们使用8个集群节点，配备1 Gbit以太网，每个节点有2个Intel Xeon E5-2630，64 GB RAM。

/* join perf */
import scala.io.Source
import scala.math._
import org.apache.spark._
import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._
import org.apache.spark.SparkConf
import scala.util.control.Breaks._
import scala.collection.mutable._
import org.apache.spark.rdd._
import org.apache.spark.sql._
import scala.util.Random
import org.apache.spark.util.SizeEstimator
import org.apache.spark.HashPartitioner

object joinPerf {

    def get_array(n: Int): Array[Int] = {
        var res = Array[Int]()
        for (x <- 1 to n) {
            res :+= Random.nextInt
        }

        return res
    }

    def main(args: Array[String]) {
        val start_time = System.nanoTime
        val conf = new SparkConf().setAppName("joinPerf")
        val sc = new SparkContext(conf)
        val cores_total = sc.getConf.get("spark.cores.max").toInt
        val partitions_total = sc.getConf.get("spark.default.parallelism").toInt
        val sqlContext = new org.apache.spark.sql.SQLContext(sc)
        import sqlContext._
        import sqlContext.implicits._
        println("start")
        val elems_total = 1024 * 1024 * 16 * cores_total
        val start_cnt = 1024 * 1024
        Random.setSeed(785354)

        var vals = Vector[Int]()
        for (x <- 1 to start_cnt) {
            vals :+= Random.nextInt
        }

        var test_rdd = sc.parallelize(vals)
        println(test_rdd.count)
        test_rdd = test_rdd.flatMap(x => get_array(elems_total / start_cnt)).distinct

        println("test_rdd count = " + test_rdd.count)
        println("partitions count = " + test_rdd.getNumPartitions)

        var test_rdd1 = test_rdd.map(x => (x, 1)).toDF("a", "b").repartition(partitions_total, $"a").cache
        var test_rdd2 = test_rdd.map(x => (x, 2)).toDF("a", "c").repartition(partitions_total, $"a").cache

        println("test_rdd1 count = " + test_rdd1.count)
        println("test_rdd2 count = " + test_rdd2.count)

        var start_test_time = System.nanoTime
        var test_res = test_rdd1.join(test_rdd2, test_rdd1("a") === test_rdd2("a"))
        println(test_res.count)
        print("join time = ")
        println((System.nanoTime - start_test_time) / 1e9d + " sec. ")

        print("all time = ")
        println((System.nanoTime - start_time) / 1e9d + " sec. ")
        sc.stop()
    }
}

配置参数：

spark.serializer                 org.apache.spark.serializer.KryoSerializer
spark.kryoserializer.buffer.max  1024
spark.kryo.unsafe                true
spark.kryo.referenceTracking     false
spark.driver.memory              22g
spark.executor.memory            22g
spark.driver.maxResultSize       22g
spark.rpc.message.maxSize        2047
spark.memory.fraction            0.8
spark.memory.storageFraction     0.5
spark.executor.extraJavaOptions  "-XX:+UseParallelGC"

每个核心的分区 - 4。

启动程序示例：

./bin/spark-submit --class "joinPerf" --conf spark.executor.cores=8 --conf spark.cores.max=64 --conf spark.default.parallelism=256 ./joinPerf.jar

理论上，随着数据大小和核心增加64倍，执行时间应该保持不变，但执行时间略有增长

不应该。虽然人们可以期望线性可扩展性，假设没有 IO 瓶颈，但当对均匀分布的数据执行严格的本地操作时，情况不再是这样，当转换需要数据交换时（RDD洗牌，Dataset Exchange）。在广泛的变革中，joins属于最昂贵的类别（下一个groupByKey-类似操作），由于它们的非还原性质，以及使用大型、本地、支持集合。

Shuffle 不仅具有高于线性的复杂度（至少O(N 日志 N）对于基于排序的方法），但也会导致数据分布不均匀，并且需要大量的磁盘和网络 IO。

如果您的代码将数据打乱两次 - 一次重新分区，情况会更加严重RDDs并一次到join Datasets (HashPartitioner for RDDs不兼容Dataset分区）。

最后，增加集群大小也会对性能产生影响，这与通信和同步开销的增加以及数据局部性的降低有关。

总体而言，您很少会看到真正的线性可扩展性，即使看到了，您也可以预期斜率

顺便说一句，我不会依赖cache - count与人一起工作时的惯用语Datasets. 可能不可靠 http://apache-spark-developers-list.1001551.n3.nabble.com/Will-count-always-trigger-an-evaluation-of-each-row-td21018.html#a21019.

也可以看看Spark：扩展核心数量时性能数据不一致 https://stackoverflow.com/q/41090127/6910411