unit-testing - 如何加速 Spark SQL 单元测试？

Question

我正在评估 Spark SQL 以实现一个简单的报告模块(对已经存储在 HDFS 上的 Avro 数据进行一些简单的聚合)。我毫不怀疑 Spark SQL 可以很好地满足我的功能和非功能需求。

但是，除了生产要求之外，我还想确保该模块是可测试的。我们遵循 BDD 方法和非常专注的场景，这意味着该模块将需要对一些非常简单的数据(1..10 条记录)运行数十/数百个 SQL 查询。

为了大致了解本地模式下 Spark SQL 的性能，我快速制作了一些测试原型(prototype):

1. 从 myTable 中选择 count(*)
2. select key, count(*) from myTable group by key

第一次测试平均耗时 100 毫秒，而第二次测试平均耗时 500 毫秒。这样的性能是 Not Acceptable ，因为它会使测试套件太慢。

为了比较，我可以使用 Crunch 及其 MemPipeline 在 10 毫秒内运行相同的测试(在本地模式下使用 MRPipeline 需要 1500 毫秒)，在嵌入式模式下使用 Hive 也需要 1500 毫秒。因此，Spark SQL 在本地模式下比 MR 快一点，但构建好的测试套件仍然慢一些。

是否可以在本地模式下加速 Spark SQL？

是否有更好/更快的方法来测试 Spark SQL 模块？

(我还没有分析执行情况，但由于 RDD 上的 groupBy().countByKey() 平均需要 40 毫秒，我希望发现罪魁祸首是查询优化器)

---

我的快速测试代码如下:

  SparkConf sparkConf = new SparkConf()
                .setMaster("local[4]")
                .setAppName("poc-sparksql");

  try (JavaSparkContext ctx = new JavaSparkContext(sparkConf)) {
        SQLContext sqlCtx = new SQLContext(ctx);

        for (int i = 0; i < ITERATIONS; i++) {
            Stopwatch testCaseSw = new Stopwatch().start();

            DataFrame df = sqlCtx.load("/tmp/test.avro", "com.databricks.spark.avro");
            df.registerTempTable("myTable");
            DataFrame result = sqlCtx.sql("select count(*) from myTable");

            System.out.println("Results: " + result.collectAsList());
            System.out.println("Elapsed: " + testCaseSw.elapsedMillis());
        }

        for (int i = 0; i < ITERATIONS; i++) {
            Stopwatch testCaseSw = new Stopwatch().start();

            DataFrame df = sqlCtx.load("/tmp/test.avro", "com.databricks.spark.avro");
            df.registerTempTable("myTable");
            DataFrame result = sqlCtx.sql("select a, count(*) from myTable group by a ");

            System.out.println("Results: " + result.collectAsList());
            System.out.println("Elapsed: " + testCaseSw.elapsedMillis());
        }
 }

Answer 1

当数据量非常小时，启动太多任务不是一个好的选择。
在您的第二个选项中，group by 将创建另一个 stage 和 200 tasks，因为你没有设置 shuffle partitions 属性，默认情况下它是 200，其中大部分都是空的。

它可能不会在单个测试中产生影响，但当您有数千个带有混洗操作的测试时，它可能会产生重大影响。

在 spark conf 中将 "spark.sql.shuffle.partitions" 设置为 x(其中 x 是 local[x])。

实际上您不需要 4 个执行程序 来处理少于 10 条记录，因此最好将执行程序的数量减少到 1 并设置 shuffle .paritions 到 1。