Spark 数据集 - 读取 CSV 并写入空输出

Question

我有一个输入文件test-reading.csv

id,sku,price
"100002701--425370728",100002701,12159
"100002701--510892030",100002701,11021
"100002701-235195215",100002701,12330
"100002701-110442364",100002701,9901
"100002701-1963746094",100002701,11243

我

编写了以下源代码，以便为我面临的问题提供一个最小、完整和可验证的示例。

有一个用于读取 CSV 文件的 ReadingRecord 类和一个用于写入输出的WritingRecord。顺便说一下，现在它们几乎相同，但在实际程序中却大不相同，因为它们表示输入和输出结构。

其余代码启动 Spark，读取 CSV，将ReadingRecord映射到WritingRecord并编写输出 CSV。

问题是：为什么如果我将for循环取消注释到flatMapGroups方法中，这个 Spark 程序会停止写入 CSV 输出？

case class ReadingRecord(var id: String, var sku: Integer, var price: Integer) {
  def toWritingRecord(): WritingRecord = {
    new WritingRecord(this.id, this.sku, this.price)
  }
}
case class WritingRecord(var id: String, var sku: Integer, var price: Integer)
object ReadingRecordEncoders {
  implicit def ReadingRecordEncoder: org.apache.spark.sql.Encoder[ReadingRecord] =
    org.apache.spark.sql.Encoders.kryo[ReadingRecord]
}
object WritingTest {
  def main(args: Array[String]) {
    val conf = new SparkConf()
      .setMaster("local[8]")
      .setAppName("writing-test")
      .set("spark.executor.memory", "1gb")
      .set("spark.num.executors", "8")
      .set("spark.executor.heartbeatInterval", "120")
    val spark = SparkSession.builder().config(conf).getOrCreate()
    import spark.implicits._
    import ReadingRecordEncoders._
    val data = spark.read.option("header", "true")
      .option("delimiter", ",")
      .option("inferSchema", "true")
      .csv("test-reading.csv")
      .map(r => {
        println(r)
        new ReadingRecord(r(0).asInstanceOf[String], r(1).asInstanceOf[Integer], r(2).asInstanceOf[Integer])
      }).groupByKey(r1 => r1.sku)
    val data1 = data.flatMapGroups((a: Integer, b: Iterator[ReadingRecord]) => {
      var list = new ArrayList[ReadingRecord]
      try {
        //        for (o <- b) {
        //          list.add(o)
        //        }
      } finally {
        list.clear()
        list = null
      }
      b.map(f => f.toWritingRecord)
    })
    data1.printSchema()
    data1.write
      .format("csv")
      .option("header", "true")
      .save("output.csv")
  }
}

Answer 1

包含注释掉的代码后，您正在尝试重用Iterator b。 使用Iterator时会对其进行修改：

特别重要的是要注意，除非另有说明，否则在调用方法后，永远不应该使用迭代器。

请参阅迭代器文档。