SPARK - 使用 RDD.foreach 创建数据框并在数据框上执行操作

Question

我是 SPARK 的新手，正在寻找更好的方法来实现以下场景。 有一个包含 3 个字段的数据库表 - 类别、金额、数量。 首先，我尝试从数据库中提取所有不同的类别。

 val categories:RDD[String] = df.select(CATEGORY).distinct().rdd.map(r => r(0).toString)

现在对于每个类别，我想执行 Pipeline，它基本上从每个类别创建数据帧并应用一些机器学习。

 categories.foreach(executePipeline)
 def execute(category: String): Unit = {
   val dfCategory = sqlCtxt.read.jdbc(JDBC_URL,"SELECT * FROM TABLE_NAME WHERE CATEGORY="+category)
dfCategory.show()    
}

有可能做这样的事情吗？还是有更好的选择？

Answer 1

// You could get all your data with a single query and convert it to an rdd
val data = sqlCtxt.read.jdbc(JDBC_URL,"SELECT * FROM TABLE_NAME).rdd

// then group the data by category
val groupedData = data.groupBy(row => row.getAs[String]("category"))

// then you get an RDD[(String, Iterable[org.apache.spark.sql.Row])]
// and you can iterate over it and execute your pipeline
groupedData.map { case (categoryName, items) =>
  //executePipeline(categoryName, items)
}

Answer 2

您的代码将因TaskNotSerializable 异常而失败，因为您尝试在execute 方法中使用SQLContext（不可序列化），该方法应被序列化并发送给工作人员以在其上执行categories RDD 中的每条记录。

假设您知道类别的数量是有限，这意味着类别列表不会太大而无法容纳您的驱动程序记忆，您应该将类别收集到驱动程序，并使用foreach 迭代该本地集合：

val categoriesRdd: RDD[String] = df.select(CATEGORY).distinct().rdd.map(r => r(0).toString)
val categories: Seq[String] = categoriesRdd.collect()
categories.foreach(executePipeline)

另一个改进是重用您加载的数据框，而不是执行另一个查询，对每个类别使用 过滤器：

def executePipeline(singleCategoryDf: DataFrame) { /* ... */ }

categories.foreach(cat => {
  val filtered = df.filter(col(CATEGORY) === cat)
  executePipeline(filtered)
})

注意：为确保 df 的重复使用不会在每次执行时重新加载它，请确保在收集类别之前cache() 它。