如何在Spark中将RDD对象转换为DataFrame

Question

我该如何将一个RDD（org.apache.spark.rdd.RDD[org.apache.spark.sql.Row]）转换为一个Dataframe（org.apache.spark.sql.DataFrame）？我使用.rdd将Dataframe转换为RDD，处理后我希望将其转换回Dataframe。我应该怎么做？

- user568109

---

在Spark 2.x中实现这一点的方法 - mrsrinivas

Answer 1

这段代码在Spark 2.x中使用Scala 2.11可以完美运行 导入必要的类。

import org.apache.spark.sql.{Row, SparkSession}
import org.apache.spark.sql.types.{DoubleType, StringType, StructField, StructType}

创建SparkSession对象，这里是spark。

val spark: SparkSession = SparkSession.builder.master("local").getOrCreate
val sc = spark.sparkContext // Just used to create test RDDs

让我们使用 RDD 将其转换为 DataFrame

val rdd = sc.parallelize(
  Seq(
    ("first", Array(2.0, 1.0, 2.1, 5.4)),
    ("test", Array(1.5, 0.5, 0.9, 3.7)),
    ("choose", Array(8.0, 2.9, 9.1, 2.5))
  )
)

##方法1 使用SparkSession.createDataFrame(RDD对象)。

val dfWithoutSchema = spark.createDataFrame(rdd)

dfWithoutSchema.show()
+------+--------------------+
|    _1|                  _2|
+------+--------------------+
| first|[2.0, 1.0, 2.1, 5.4]|
|  test|[1.5, 0.5, 0.9, 3.7]|
|choose|[8.0, 2.9, 9.1, 2.5]|
+------+--------------------+

##方法2 使用SparkSession.createDataFrame(RDD对象)并指定列名。

val dfWithSchema = spark.createDataFrame(rdd).toDF("id", "vals")

dfWithSchema.show()
+------+--------------------+
|    id|                vals|
+------+--------------------+
| first|[2.0, 1.0, 2.1, 5.4]|
|  test|[1.5, 0.5, 0.9, 3.7]|
|choose|[8.0, 2.9, 9.1, 2.5]|
+------+--------------------+

##方法3（问题的实际答案） 这种方法需要输入rdd应该是类型为RDD [Row]。

val rowsRdd: RDD[Row] = sc.parallelize(
  Seq(
    Row("first", 2.0, 7.0),
    Row("second", 3.5, 2.5),
    Row("third", 7.0, 5.9)
  )
)

创建模式

val schema = new StructType()
  .add(StructField("id", StringType, true))
  .add(StructField("val1", DoubleType, true))
  .add(StructField("val2", DoubleType, true))

现在将rowsRdd 和 schema 应用到 createDataFrame()函数中。

val df = spark.createDataFrame(rowsRdd, schema)

df.show() 
+------+----+----+
|    id|val1|val2|
+------+----+----+
| first| 2.0| 7.0|
|second| 3.5| 2.5|
| third| 7.0| 5.9|
+------+----+----+

- mrsrinivas

---

感谢您以易懂的方式展示了使用createDataFrame的不同方法。 - anaotha

第三种方法在数据块上很有用，其他方法都无法正常工作并且会出现错误。 - Narendra Maru

Answer 2

SparkSession有多个createDataFrame方法，可以基于一个RDD创建一个DataFrame。我想其中之一应该可以适用于您的上下文。

例如：

def createDataFrame(rowRDD: RDD[Row], schema: StructType): DataFrame

使用给定的模式，从包含行的RDD创建一个DataFrame。

- The Archetypal Paul

Answer 3

假设您的RDD [row] 名为 rdd，您可以使用：

val sqlContext = new SQLContext(sc) 
import sqlContext.implicits._
rdd.toDF()

- dtjones

---

我认为它不适用于RDD[Row]。我有什么遗漏吗？ - Daniel de Paula

自 Spark 2.0 起，SQLContext 已被 SparkSession 取代，但为了向后兼容，该类在代码库中保留（scaladoc）。使用它会产生弃用警告。 - tomaskazemekas

这个不起作用。RDD [Row] 不提供该选项。需要将其转换为 case class RDD 的元组。 - sho

Answer 4

注意：此答案最初发布在这里

我发布这个答案是因为我想分享其他答案中没有发现的可用选项的更多细节

---

从Rows的RDD创建DataFrame有两个主要选项：

1) 如已指出，您可以使用toDF()，该函数可以通过import sqlContext.implicits._导入。但是，此方法仅适用于以下类型的RDD：

• RDD[Int]
• RDD[Long]
• RDD[String]
• RDD[T <: scala.Product]

(来源：SQLContext.implicits对象的 Scaladoc)

最后一个签名实际上意味着它可以用于元组的RDD或案例类的RDD（因为元组和案例类都是scala.Product的子类）。

因此，要将此方法用于RDD[Row]，您必须将其映射到RDD[T <: scala.Product]。这可以通过将每一行映射到自定义案例类或元组来完成，就像以下示例代码中所示：

val df = rdd.map({ 
  case Row(val1: String, ..., valN: Long) => (val1, ..., valN)
}).toDF("col1_name", ..., "colN_name")

或者

case class MyClass(val1: String, ..., valN: Long = 0L)
val df = rdd.map({ 
  case Row(val1: String, ..., valN: Long) => MyClass(val1, ..., valN)
}).toDF("col1_name", ..., "colN_name")

这种方法的主要缺点（在我看来）是你需要在map函数中逐列明确设置结果DataFrame的模式。如果你事先不知道模式，可能可以通过编程方式完成，但那里可能会有些混乱。因此，另一种选择是：

2）您可以使用createDataFrame（rowRDD：RDD [Row]，schema：StructType），如已接受的答案中所述，该选项可在SQLContext对象中使用。以下是将旧DataFrame的RDD转换为示例：

val rdd = oldDF.rdd
val newDF = oldDF.sqlContext.createDataFrame(rdd, oldDF.schema)

请注意，不需要显式设置任何模式列。我们复用旧的DF模式，该模式属于 StructType 类，并且可以轻松扩展。但是，这种方法有时不可行，在某些情况下可能比第一种方法效率低。

- Daniel de Paula

---

感谢您提供详细的 import sqlContext.implicits.。 - WestCoastProjects

未来请不要在多个问题中发布相同的答案。如果这些问题是重复的，请发表一个好的答案，然后投票或标记关闭其他问题作为重复。如果问题不是重复，则根据问题调整您的答案。请参阅如何编写好的答案？。 - user3956566

Answer 5

假设你有一个DataFrame，想要对字段数据进行一些修改，通过将其转换为RDD[Row]实现。

val aRdd = aDF.map(x=>Row(x.getAs[Long]("id"),x.getAs[List[String]]("role").head))

要将RDD转换回DataFrame，我们需要定义RDD的结构类型。

如果数据类型为Long，则在结构中它将变为LongType。

如果是String，则在结构中为StringType。

val aStruct = new StructType(Array(StructField("id",LongType,nullable = true),StructField("role",StringType,nullable = true)))

现在您可以使用createDataFrame方法将RDD转换为DataFrame。

val aNamedDF = sqlContext.createDataFrame(aRdd,aStruct)

- Ajay Gupta

Answer 6

方法一：（Scala）

val sqlContext = new org.apache.spark.sql.SQLContext(sc)
import sqlContext.implicits._
val df_2 = sc.parallelize(Seq((1L, 3.0, "a"), (2L, -1.0, "b"), (3L, 0.0, "c"))).toDF("x", "y", "z")

方法2：（Scala）

case class temp(val1: String,val3 : Double) 

val rdd = sc.parallelize(Seq(
  Row("foo",  0.5), Row("bar",  0.0)
))
val rows = rdd.map({case Row(val1:String,val3:Double) => temp(val1,val3)}).toDF()
rows.show()

方法1：（Python）

from pyspark.sql import Row
l = [('Alice',2)]
Person = Row('name','age')
rdd = sc.parallelize(l)
person = rdd.map(lambda r:Person(*r))
df2 = sqlContext.createDataFrame(person)
df2.show()

方法二：（Python）

from pyspark.sql.types import * 
l = [('Alice',2)]
rdd = sc.parallelize(l)
schema =  StructType([StructField ("name" , StringType(), True) , 
StructField("age" , IntegerType(), True)]) 
df3 = sqlContext.createDataFrame(rdd, schema) 
df3.show()

从行对象中提取值，然后应用案例类将rdd转换为DF。

val temp1 = attrib1.map{case Row ( key: Int ) => s"$key" }
val temp2 = attrib2.map{case Row ( key: Int) => s"$key" }

case class RLT (id: String, attrib_1 : String, attrib_2 : String)
import hiveContext.implicits._

val df = result.map{ s => RLT(s(0),s(1),s(2)) }.toDF

- Aravind Krishnakumar

Answer 7

这是将您的列表转换为Spark RDD，然后将该Spark RDD转换为Dataframe的简单示例。

请注意，我使用了Spark-shell的scala REPL来执行以下代码，其中sc是SparkContext的一个实例，它在Spark-shell中隐式可用。希望它能回答您的问题。

scala> val numList = List(1,2,3,4,5)
numList: List[Int] = List(1, 2, 3, 4, 5)

scala> val numRDD = sc.parallelize(numList)
numRDD: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[80] at parallelize at <console>:28

scala> val numDF = numRDD.toDF
numDF: org.apache.spark.sql.DataFrame = [_1: int]

scala> numDF.show
+---+
| _1|
+---+
|  1|
|  2|
|  3|
|  4|
|  5|
+---+

- Rashmit Rathod

---

一个有趣的事实是：当你的列表是Double类型而不是int（或Long，String，<：Product）时，这个功能就会停止工作。 - Rick Moritz

不回答 OP：它谈到了 RDD[Row]。 - WestCoastProjects

Answer 8

在较新版本的Spark（2.0+）中：

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

val spark = SparkSession
  .builder()
  .getOrCreate()
import spark.implicits._

val dfSchema = Seq("col1", "col2", "col3")
rdd.toDF(dfSchema: _*)

- ozzieisaacs

---

sparkSession只是sqlContext和hiveContext的包装器。 - Archit

Answer 9

我会尝试使用“字数问题”来解释解决方案。 1. 使用sc读取文件。

2. Produce word count

3. Methods to create DF

   • rdd.toDF method
   • rdd.toDF("word","count")
     • spark.createDataFrame(rdd,schema)

   Read file using spark

   val rdd=sc.textFile("D://cca175/data/")  
   

   Rdd to Dataframe

   val df=sc.textFile("D://cca175/data/").toDF("t1") df.show

   Method 1

   Create word count RDD to Dataframe

   val df=rdd.flatMap(x=>x.split(" ")).map(x=>(x,1)).reduceByKey((x,y)=>(x+y)).toDF("word","count")
   

   Method2

   Create Dataframe from Rdd

   val df=spark.createDataFrame(wordRdd) 
   # with header   
   val df=spark.createDataFrame(wordRdd).toDF("word","count")  df.show
   

   Method3

   Define Schema

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

   val schema=new StructType(). add(StructField("word",StringType,true)). add(StructField("count",StringType,true))

   Create RowRDD

   import org.apache.spark.sql.Row
   val rowRdd=wordRdd.map(x=>(Row(x._1,x._2)))     
   

   Create DataFrame from RDD with schema

   val df=spark.createDataFrame(rowRdd,schema)
   df.show

- Priyanshu Singh

Answer 10

我遇到了同样的问题，最终解决了。这很简单易行。

1. 你需要添加这段代码 import sc.implicits._，其中sc表示SQLContext。添加此代码后，你将获得rdd.toDF()方法。
2. 将rdd[RawData]转换为rdd[YourCaseClass]。例如，你有一个rdd类型如下：rdd[(String, Integer, Long)]，你可以创建一个Case Class YourCaseClass(name: String, age: Integer, timestamp: Long)并将原始rdd转换为具有YourCaseClass类型的rdd，然后你就会得到rdd[YourCaseClass]
3. 将rdd[YourCaseClass]保存到Hive表中。使用Case Class来表示rdd类型，我们可以避免命名每个列字段或与StructType相关的模式。yourRdd.toDF().write.format("parquet").mode(SaveMode.Overwrite).insertInto(yourHiveTableName)

- Armstrongya