使用 Spark 数据帧进行单元测试答案

【问题标题】：Unit testing with Spark dataframes使用 Spark 数据帧进行单元测试
【发布时间】：2016-07-03 19:09:28
【问题描述】：

我正在尝试测试对数据帧执行转换的程序的一部分我想测试这些数据帧的几种不同变体，这些变体排除了从文件中读取特定 DF 的选项

所以我的问题是：

有没有关于如何使用 Spark 和数据帧执行单元测试的好教程，尤其是关于数据帧的创建？
如何在没有大量样板文件且不从文件中读取这些内容的情况下创建这些不同的几行数据帧？
是否有任何实用程序类用于检查数据框中的特定值？

显然我之前用谷歌搜索过，但找不到任何非常有用的东西。我发现的更有用的链接包括：

如果示例/教程在 Scala 中，那就太好了，但我会采用你所掌握的任何语言

提前致谢

【问题讨论】：

这太离题了 :) 但好的起点是查看the Spark source。

标签： scala unit-testing apache-spark apache-spark-sql spark-dataframe

【解决方案1】：

这个link 展示了我们如何以编程方式创建带有模式的数据框。您可以将数据保存在单独的特征中，并将其与您的测试混合。例如，

// This example assumes CSV data. But same approach should work for other formats as well.

trait TestData {
  val data1 = List(
    "this,is,valid,data",
    "this,is,in-valid,data",
  )
  val data2 = ...  
}

然后使用 ScalaTest，我们可以做这样的事情。

class MyDFTest extends FlatSpec with Matchers {

  "method" should "perform this" in new TestData {
     // You can access your test data here. Use it to create the DataFrame.
     // Your test here.
  }
}

要创建 DataFrame，您可以使用以下几个 util 方法。

  def schema(types: Array[String], cols: Array[String]) = {
    val datatypes = types.map {
      case "String" => StringType
      case "Long" => LongType
      case "Double" => DoubleType
      // Add more types here based on your data.
      case _ => StringType
    }
    StructType(cols.indices.map(x => StructField(cols(x), datatypes(x))).toArray)
  }

  def df(data: List[String], types: Array[String], cols: Array[String]) = {
    val rdd = sc.parallelize(data)
    val parser = new CSVParser(',')
    val split = rdd.map(line => parser.parseLine(line))
    val rdd = split.map(arr => Row(arr(0), arr(1), arr(2), arr(3)))
    sqlContext.createDataFrame(rdd, schema(types, cols))
  }

我不知道有任何实用程序类可以检查 DataFrame 中的特定值。但我认为使用 DataFrame API 编写一个应该很简单。

【讨论】：

谢谢，这对我来说似乎已经足够了。另外查看 Spark 源代码也给了我一些想法。谢谢！

【解决方案2】：

对于那些希望在 Java 中实现类似功能的人，您可以使用 start by using this project 在单元测试中初始化 SparkContext：https://github.com/holdenk/spark-testing-base

我个人不得不模仿一些 AVRO 文件的文件结构。所以我使用 Avro-tools (https://avro.apache.org/docs/1.8.2/gettingstartedjava.html#download_install) 使用以下命令从我的二进制记录中提取模式：

java -jar $AVRO_HOME/avro tojson largeAvroFile.avro | head -3

然后，使用这个小助手方法，您可以将输出 JSON 转换为 DataFrame 以在您的单元测试中使用。

private DataFrame getDataFrameFromList() {
    SQLContext sqlContext = new SQLContext(jsc());
    ImmutableList<String> elements = ImmutableList.of(
        {"header":{"appId":"myAppId1","clientIp":"10.22.63.3","createdDate":"2017-05-10T02:09:59.984Z"}}
        {"header":{"appId":"myAppId1","clientIp":"11.22.63.3","createdDate":"2017-05-11T02:09:59.984Z"}}
        {"header":{"appId":"myAppId1","clientIp":"12.22.63.3","createdDate":"2017-05-11T02:09:59.984Z"}}
    );
    JavaRDD<String> parallelize = jsc().parallelize(elements);
    return sqlContext.read().json(parallelize);
}

【讨论】：

【解决方案3】：

您可以使用 Spark 用于其自己的单元测试的 SharedSQLContext 和 SharedSparkSession。查看我的answer 以获取示例。

【讨论】：