在地图中创建 Spark Row答案

【问题标题】：Create Spark Row in a map在地图中创建 Spark Row
【发布时间】：2016-09-23 20:57:43
【问题描述】：

我在https://databricks.com/blog/2015/02/17/introducing-dataframes-in-spark-for-large-scale-data-science.html 看到了一个用 Python 编写的 Dataframes 教程。我正在尝试将其翻译成 Scala。

他们有以下代码：

df = context.load("/path/to/people.json")
# RDD-style methods such as map, flatMap are available on DataFrames
# Split the bio text into multiple words.
words = df.select("bio").flatMap(lambda row: row.bio.split(" "))
# Create a new DataFrame to count the number of words
words_df = words.map(lambda w: Row(word=w, cnt=1)).toDF()
word_counts = words_df.groupBy("word").sum()

所以，我首先将csv 中的数据读入数据框df，然后我有：

val title_words = df.select("title").flatMap { row =>    
  row.getAs[String("title").split(" ") }
val title_words_df = title_words.map( w => Row(w,1) ).toDF()
val word_counts = title_words_df.groupBy("word").sum()

但我不知道：

如何将字段名称分配给以 val title_words_df = ... 开头的行中的行
我遇到错误“toDF 的值不是 org.apache.spark.rdd.RDD[org.apache.spark.sql.Row] 的成员”

提前感谢您的帮助。

【问题讨论】：

标签： python scala apache-spark apache-spark-sql

【解决方案1】：

如何将字段名称分配给行

Python Row 与 Scala 对应的对象类型完全不同。它是一个增加了名称的元组，这使得它比无类型集合更等同于产品类型 (o.a.s.sql.Row)。

我遇到错误“toDF 的值不是 org.apache.spark.rdd.RDD[org.apache.spark.sql.Row] 的成员”

由于o.a.s.sql.Row 基本上是无类型的，它不能与toDF 一起使用，并且需要具有显式架构的createDataFrame。

import org.apache.spark.sql.types._

val schema = StructType(Seq(
  StructField("word", StringType), StructField("cnt", LongType)
))

sqlContext.createDataFrame(title_words.map(w => Row(w, 1L)), schema)

如果您希望您的代码与 Python 版本等效，则应使用产品类型而不是 Row。这意味着Tuple:

title_words.map((_, 1L)).toDF("word", "cnt")

或案例类：

case class Record(word: String, cnt: Long)

title_words.map(Record(_, 1L)).toDF

但实际上，应该不需要使用 RDD：

import org.apache.spark.sql.functions.{explode, lit, split}

df.select(explode(split($"title", " ")), lit(1L))

【讨论】：