如何一次将每组发送给spark执行者？

Question

我无法一次将每组数据帧发送给执行器。

我在公司_模型_VAL_df数据框中有如下数据。

 ----------------------------------------------------------------------------------------
 | model_id  |  fiscal_year  | fiscal_quarter | col1 | col2 | col3 | col4 | col5 | col6 |
 ----------------------------------------------------------------------------------------
 |    1      | 2018          |   1             | r1   | r2   | r3   |  r4  | r5   |  r6 |
 |    1      | 2018          |   2             | r1   | r2   | r3   |  r4  | r5   |  r6 |
 |    1      | 2018          |   1             | r1   | r2   | r3   |  r4  | r5   |  r6 |
 |    1      | 2018          |   2             | r1   | r2   | r3   |  r4  | r5   |  r6 |
 |    1      | 2018          |   1             | r1   | r2   | r3   |  r4  | r5   |  r6 |
 |    2      | 2017          |   3             | r1   | r2   | r3   |  r4  | r5   |  r6 |
 |    2      | 2017          |   1             | r1   | r2   | r3   |  r4  | r5   |  r6 |
 |    2      | 2017          |   3             | r1   | r2   | r3   |  r4  | r5   |  r6 |
 |    2      | 2017          |   3             | r1   | r2   | r3   |  r4  | r5   |  r6 |
 |    2      | 2017          |   1             | r1   | r2   | r3   |  r4  | r5   |  r6 |
 ----------------------------------------------------------------------------------------

我想将每个分组的数据发送给executor，以便一次处理每个数据。

为此，我做了如下工作：

var dist_company_model_vals_df =  company_model_vals_df.select("model_id","fiscal_quarter","fiscal_year").distinct()

// Want to send each group at a time to write by executors.

dist_company_model_vals_df.foreach(rowDf => {
  writeAsParquet(rowDf , parquet_file)    // this simply writes the data as parquet file
})

错误：

这将抛出一个NullPointerExcure，因为在Execator端找不到rowDf。使用Scala 2.11在Spark-sql中处理此问题的正确方法是什么？

第二部分：问题

当我做公司模型时。groupBy（“model_id”、“fiscal_quarty”、“fiscal_year”）即使在我增加内存后，数据仍大量溢出到磁盘上。也就是说，公司模型是巨大的数据帧。。。做群比时会发生很多溢出。

同样是下面的情况下，即与分区通过

company_model_vals_df.write.partition（model_id、fiscal_quarter、fiscal_year）

PSEDO代码：所以为了避免是第一个，我会做valgroups=company_model_vals_df的元组。groupBy（“型号id”、“会计季度”、“会计年度”）。收集

groups.forEach{ group ->
   // I want to prepare child dataframes for each group from    company_model_vals_df

   val child_df = company_model_vals_df.where(model_id= group.model_id && fiscal_quarter === group.fiscal_quarter && etc)

 this child_df , i want wrote to a file i.e. saveAs(path)
}

无论如何都有办法做到这一点。这里有对我有用的spark函数或API吗？请提出解决方法。

Answer 1

如果我正确理解了您的问题，您希望分别处理每个“型号id”、“财政季度”、“财政年度”的数据。

如果这是正确的，您可以使用groupBy（），例如：

company_model_vals_df.groupBy("model_id","fiscal_quarter","fiscal_year").agg(avg($"col1") as "average")

如果要将每个逻辑组写入一个单独的文件夹，可以通过以下方式完成：

scala prettyprint-override">company_model_vals_df.write.partitionBy("model_id","fiscal_quarter","fiscal_year").parquet("path/to/save")

Answer 2

这里几乎没有选择-

• 您需要将数据集分成多个数据集，并分别处理它们，就像

var dist_company_model_vals_list =  company_model_vals_df
  .select("model_id","fiscal_quarter","fiscal_year").distinct().collectAsList

然后过滤company_model_vals_df，输出dist_company_model_vals_listlist，它提供了几个可以独立工作的数据集，如

def rowList = {
import org.apache.spark.sql._
var dfList:Seq[DataFrame] = Seq()
for (data <- dist_company_model_vals_list.zipWithIndex) {
val i = data._2
val row = data.-1
val filterCol = col($"model_id").equalTo(row.get(i).getInt(0).and($"fiscal_quarter").equalTo(row.get(i).getInt(1).and($"fiscal_year").equalTo(row.get(i).getInt(2))

   val resultDf = company_model_vals_df.filter(filterCol)    
dfList +: = resultDf
      }
dfList
}

• 如果您的目标是写入数据，则可以在DataFrameWriter上使用partitionBy（“model_id”、“fiscal_quarter”、“fiscal_year”）方法分别写入数据