【问题标题】：SparkSQL performance issue with collect method收集方法的 SparkSQL 性能问题
【发布时间】：2018-05-14 07:54:29
【问题描述】：

我们目前在使用 scala 语言编写的 sparksql 中遇到性能问题。申请流程如下。

Spark 应用程序从输入 hdfs 目录读取文本文件
使用以编程方式指定架构在文件顶部创建数据框。该数据帧将是保存在内存中的输入文件的精确复制。数据框中将有大约 18 列

var eqpDF = sqlContext.createDataFrame(eqpRowRdd, eqpSchema)
从第 2 步中构建的第一个数据框创建过滤后的数据框。此数据框将在不同关键字的帮助下包含唯一的帐号。

var distAccNrsDF = eqpDF.select("accountnumber").distinct().collect()
使用第2步和第3步构建的两个数据框，我们将获取属于一个帐号的所有记录，并在过滤后的数据之上做一些Json解析逻辑。

var filtrEqpDF = eqpDF.where("accountnumber='" + data.getString(0) + "'").collect()
最后将json解析后的数据放入Hbase表中

在这里，我们在数据帧上调用 collect 方法时遇到了性能问题。因为 collect 会将所有数据提取到单个节点中然后进行处理，从而失去了并行处理的好处。同样在实际场景中，我们可以预期会有 100 亿条数据记录。因此，将所有这些记录收集到驱动程序节点可能会由于内存或磁盘空间限制而导致程序本身崩溃。

我认为在我们的案例中不能使用 take 方法，它一次只能获取有限数量的记录。我们必须从整个数据中获取所有唯一帐号，因此我不确定是否采取方法，该方法采用一次有限的记录，将满足我们的要求

感谢您对避免调用 collect 方法的任何帮助，并有一些其他的最佳实践可供遵循。如果有人遇到过类似问题，代码 sn-ps/suggestions/git 链接将非常有帮助

代码 sn-p

    val eqpSchemaString = "acoountnumber ....."
  val eqpSchema = StructType(eqpSchemaString.split(" ").map(fieldName => 
StructField(fieldName, StringType, true))); 
    val eqpRdd = sc.textFile(inputPath)
    val eqpRowRdd = eqpRdd.map(_.split(",")).map(eqpRow => Row(eqpRow(0).trim, eqpRow(1).trim, ....)

    var eqpDF = sqlContext.createDataFrame(eqpRowRdd, eqpSchema);


    var distAccNrsDF = eqpDF.select("accountnumber").distinct().collect()


    distAccNrsDF.foreach { data =>

      var filtrEqpDF = eqpDF.where("accountnumber='" + data.getString(0) + "'").collect()



      var result = new JSONObject()

      result.put("jsonSchemaVersion", "1.0")
      val firstRowAcc = filtrEqpDF(0)
      //Json parsing logic 
      {
         .....
         .....
      }
    }

【问题讨论】：

你真正想做什么？只需写入 Hbase 表？如果是这样，您为什么要收集或拍摄？ collect 和 take 仅用于查看样本数据。除此之外，无需使用 collect 或 take。
基本上我们需要将属于同一个账号的所有数据（在源文件中）进行分组，并且必须将分组的数据推送到hbase。收集我们使用的是因为，以便在全球范围内找出不同的帐号。如果不调用collect，如何全局找出可能分布在多个节点的唯一账号。

标签： scala apache-spark spark-dataframe

【解决方案1】：

这种情况下通常采取的做法是：

调用foreachPartition，而不是collect：foreachPartition 将函数分别应用于底层DataFrame 的每个分区（由Iterator[Row] 表示）（分区是Spark 并行性的原子单元）
该函数将打开一个到 HBase 的连接（从而使每个分区一个）并通过该连接发送所有包含的值

这意味着每个执行器都会打开一个连接（它不可序列化但位于函数的边界内，因此不需要通过网络发送）并将其内容独立发送到 HBase，无需收集所有数据在驱动程序上（或任何一个节点，就此而言）。

看起来您正在读取一个 CSV 文件，因此可能类似以下内容可以解决问题：

spark.read.csv(inputPath).         // Using DataFrameReader but your way works too
  foreachPartition { rows =>
    val conn = ???                 // Create HBase connection
    for (row <- rows) {            // Loop over the iterator
      val data = parseJson(row)    // Your parsing logic
      ???                          // Use 'conn' to save 'data'
    }
  }

【讨论】：

【解决方案2】：

如果您有大量数据，您可以忽略代码中的收集。

Collect 在驱动程序中将数据集的所有元素作为数组返回。这通常在过滤器或其他返回足够小的数据子集的操作之后很有用。

不过，这也可能导致驱动程序耗尽内存，因为 collect() 将整个 RDD/DF 获取到单个机器。

我刚刚编辑了你的代码，应该对你有用。

        var distAccNrsDF = eqpDF.select("accountnumber").distinct()
            distAccNrsDF.foreach { data =>
              var filtrEqpDF = eqpDF.where("accountnumber='" + data.getString(0) + "'")
              var result = new JSONObject()
              result.put("jsonSchemaVersion", "1.0")
              val firstRowAcc = filtrEqpDF(0)
              //Json parsing logic 
              {
                 .....
                 .....
              }
            }

【讨论】：