在地图中创建Spark Row

Question

我在网上看到了Dataframes教程https://databricks.com/blog/2015/02/17/introducing-dataframes-in-spark-for-large-scale-data-science.html这是用Python编写的。我正试图把它翻译成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()

但我不知道：

>

如何将字段名分配给以valtitle_words_df=...

我有错误的值toDF不是一个成员org.apache.spark.rdd.RDD[org.apache.spark.sql.行]

提前谢谢你的帮助。

Answer 1

如何为行分配字段名

PythonRow与Scala对应的对象类型截然不同。它是一个用名称扩充的元组，与非类型集合（o.a.s.sql.Row）相比，它更等同于产品类型。

我有错误的值toDF不是一个成员org.apache.spark.rdd.RDD[org.apache.spark.sql.行]

因为o.a.s.sql。行基本上是非类型化的，它不能与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))