如何在火花中将rdd对象转换为数据框

Posted

技术标签:

【中文标题】如何在火花中将rdd对象转换为数据框【英文标题】：How to convert rdd object to dataframe in spark 【发布时间】：2015-04-01 05:38:33 【问题描述】：

如何将 RDD (org.apache.spark.rdd.RDD[org.apache.spark.sql.Row]) 转换为 Dataframe org.apache.spark.sql.DataFrame。我使用.rdd 将数据帧转换为rdd。处理后我希望它回到数据框中。我该怎么做？

【问题讨论】：

在Spark 2.x中实现这一点的方法 【参考方案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 obj)。

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]|
+------+--------------------+

##方法二 使用SparkSession.createDataFrame(RDD obj) 并指定列名。

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|
+------+----+----+

【讨论】：

感谢您以易于理解的方式展示使用 createDataFrame 的不同方式 第三种方法对数据砖很有帮助，因为其他方法不工作并给出错误【参考方案2】：

SparkSession 有许多 createDataFrame 方法，在给定 RDD 的情况下创建 DataFrame。我想其中之一将适用于您的上下文。

例如：

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

使用给定的 RDD 从包含 Rows 的 RDD 创建一个 DataFrame 架构。

【讨论】：

【参考方案3】：

假设你的 RDD[row] 被称为 rdd，你可以使用：

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

【讨论】：

我认为它不适用于 RDD[Row]。我错过了什么吗？ 由于 Spark 2.0 SQLContext 被 SparkSession 取代，但该类保留在代码库中以实现向后兼容性(scaladoc). 使用它会引发弃用警告。【参考方案4】：

注意：这个答案最初发布于here

我发布此答案是因为我想分享有关我在其他答案中没有找到的可用选项的更多详细信息

要从 RDD 的 Rows 创建 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]。这可以通过将每一行映射到自定义案例类或元组来完成，如以下代码 sn-ps：

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 类，并且可以轻松扩展。但是，这种方法有时是行不通的，在某些情况下可能不如第一种方法有效。

【讨论】：

感谢详细信息import sqlContext.implicits. 以后，请不要对多个问题发布相同的答案。如果问题是重复的，请发布一个好的答案，然后投票或标记以关闭另一个问题作为重复。如果问题不重复，调整您对该问题的回答。请参阅How do I write a good 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)

【讨论】：

【参考方案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")

方法二：(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()

方法一：（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()

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

val temp1 = attrib1.mapcase Row ( key: Int ) => s"$key" 
val temp2 = attrib2.mapcase 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

【讨论】：

【参考方案7】：

这是一个简单的例子，将你的 List 转换为 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|
+---+

【讨论】：

一个有趣的事实：当您的 List 是 Double 而不是 int（或 Long、String、<: product> 不回答 OP : 其中谈到 RDD[Row]【参考方案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: _*)

【讨论】：

sparkSession 只是 sqlContext、hiveContext 的一个包装器【参考方案9】： 
One needs to create a schema, and attach it to the Rdd.

假设 val spark 是 SparkSession.builder 的产物...

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

    /* Lets gin up some sample data:
     * As RDD's and dataframes can have columns of differing types, lets make our
     * sample data a three wide, two tall, rectangle of mixed types.
     * A column of Strings, a column of Longs, and a column of Doubules 
     */
    val arrayOfArrayOfAnys = Array.ofDim[Any](2,3)
    arrayOfArrayOfAnys(0)(0)="aString"
    arrayOfArrayOfAnys(0)(1)=0L
    arrayOfArrayOfAnys(0)(2)=3.14159
    arrayOfArrayOfAnys(1)(0)="bString"
    arrayOfArrayOfAnys(1)(1)=9876543210L
    arrayOfArrayOfAnys(1)(2)=2.71828

    /* The way to convert an anything which looks rectangular, 
     * (Array[Array[String]] or Array[Array[Any]] or Array[Row], ... ) into an RDD is to 
     * throw it into sparkContext.parallelize.
     * http://spark.apache.org/docs/latest/api/scala/index.html#org.apache.spark.SparkContext shows
     * the parallelize definition as 
     *     def parallelize[T](seq: Seq[T], numSlices: Int = defaultParallelism)
     * so in our case our ArrayOfArrayOfAnys is treated as a sequence of ArraysOfAnys.
     * Will leave the numSlices as the defaultParallelism, as I have no particular cause to change it. 
     */
    val rddOfArrayOfArrayOfAnys=spark.sparkContext.parallelize(arrayOfArrayOfAnys)

    /* We'll be using the sqlContext.createDataFrame to add a schema our RDD.
     * The RDD which goes into createDataFrame is an RDD[Row] which is not what we happen to have.
     * To convert anything one tall and several wide into a Row, one can use Row.fromSeq(thatThing.toSeq)
     * As we have an RDD[somethingWeDontWant], we can map each of the RDD rows into the desired Row type. 
     */     
    val rddOfRows=rddOfArrayOfArrayOfAnys.map(f=>
        Row.fromSeq(f.toSeq)
    )

    /* Now to construct our schema. This needs to be a StructType of 1 StructField per column in our dataframe.
     * https://spark.apache.org/docs/latest/api/scala/index.html#org.apache.spark.sql.types.StructField shows the definition as
     *   case class StructField(name: String, dataType: DataType, nullable: Boolean = true, metadata: Metadata = Metadata.empty)
     * Will leave the two default values in place for each of the columns:
     *        nullability as true, 
     *        metadata as an empty Map[String,Any]
     *   
     */

    val schema = StructType(
        StructField("colOfStrings", StringType) ::
        StructField("colOfLongs"  , LongType  ) ::
        StructField("colOfDoubles", DoubleType) ::
        Nil
    )

    val df=spark.sqlContext.createDataFrame(rddOfRows,schema)
    /*
     *      +------------+----------+------------+
     *      |colOfStrings|colOfLongs|colOfDoubles|
     *      +------------+----------+------------+
     *      |     aString|         0|     3.14159|
     *      |     bString|9876543210|     2.71828|
     *      +------------+----------+------------+
    */ 
    df.show

步骤相同，但 val 声明更少：

    val arrayOfArrayOfAnys=Array(
        Array("aString",0L         ,3.14159),
        Array("bString",9876543210L,2.71828)
    )

    val rddOfRows=spark.sparkContext.parallelize(arrayOfArrayOfAnys).map(f=>Row.fromSeq(f.toSeq))

    /* If one knows the datatypes, for instance from JDBC queries as to RDBC column metadata:
     * Consider constructing the schema from an Array[StructField].  This would allow looping over 
     * the columns, with a match statement applying the appropriate sql datatypes as the second
     *  StructField arguments.   
     */
    val sf=new Array[StructField](3)
    sf(0)=StructField("colOfStrings",StringType)
    sf(1)=StructField("colOfLongs"  ,LongType  )
    sf(2)=StructField("colOfDoubles",DoubleType)        
    val df=spark.sqlContext.createDataFrame(rddOfRows,StructType(sf.toList))
    df.show

【讨论】：

【参考方案10】：

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

    产生字数

    创建DF的方法

    rdd.toDF 方法 rdd.toDF("word","count") spark.createDataFrame(rdd,schema)

    使用 spark 读取文件

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

    Rdd 到数据框

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

    方法一

    创建字数 RDD 到 Dataframe

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

    方法2

    从 Rdd 创建数据框

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

    方法3

    定义架构

    导入 org.apache.spark.sql.types._

    val schema=new StructType()。 添加（StructField（“单词”，StringType，true））。 add(StructField("count",StringType,true))

    创建 RowRDD

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

    使用架构从 RDD 创建 DataFrame

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

【讨论】：

【参考方案11】：

我遇到了同样的问题，终于解决了。这非常简单易行。

    您必须添加此代码import sc.implicits._，sc 表示SQLContext。添加此代码您将获得 rdd.toDF() 方法。 将您的rdd[RawData] 转换为rdd[YourCaseClass]。例如，你有一个像 rdd[(String, Integer, Long)] 这样的 rdd 类型，你可以创建一个案例类 YourCaseClass(name: String, age: Integer, timestamp: Long) 并将原始 rdd 转换为 YourCaseClass 类型的 rdd，然后你得到 rdd[YourCaseClass] 保存rdd[YourCaseClass] 到蜂巢表。 yourRdd.toDF().write.format("parquet").mode(SaveMode.Overwrite).insertInto(yourHiveTableName)用case类表示rdd类型，我们可以避免命名每个列字段或StructType相关架构。

【讨论】：

【参考方案12】：

要将 Array[Row] 转换为 DataFrame 或 Dataset，以下操作很优雅：

说，schema 是行的 StructType，那么

val rows: Array[Row]=...
implicit val encoder = RowEncoder.apply(schema)
import spark.implicits._
rows.toDS

【讨论】：

以上是关于如何在火花中将rdd对象转换为数据框的主要内容，如果未能解决你的问题，请参考以下文章

如何在 Spark 中将 Avro Schema 对象转换为 StructType

Pyspark：从列表的 RDD 创建一个火花数据框，其中列表的某些元素是对象

如何在pyspark中将rdd行转换为带有json结构的数据框？

如何在火花中将列转换为数组[长]

在 Shiny 中将数据从 rhandsontable 对象转换为数据框

在 Pyspark 中将字典转换为数据框