如何将rdd对象转换为spark中的dataframe

Question

如何将RDD(org.apache.spark.rdd.RDD[org.apache.spark.sql.Row])转换为Dataframe org.apache.spark.sql.DataFrame.我使用了将数据帧转换为rdd .rdd.处理完之后我想把它放回到数据帧中.我怎样才能做到这一点 ？

Answer 1

SqlContext有许多createDataFrame方法可以创建DataFrame给定的RDD.我想其中一个将适用于您的上下文.

例如:

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

使用给定的模式从包含Rows的RDD创建DataFrame.

Answer 2

这段代码与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]|
+------+--------------------+

方法2

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

Answer 3

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

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

Answer 4

注意:这个答案最初发布在这里

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

要从RDD of Rows创建DataFrame,有两个主要选项:

1)正如已经指出的那样,你可以使用toDF()哪些可以导入import sqlContext.implicits._.但是,此方法仅适用于以下类型的RDD:

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

(来源:Scaladoc所述的SQLContext.implicits对象)

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

因此,要将此方法用于a 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对象中提供的接受答案中使用as .转换旧DataFrame的RDD的示例:

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

请注意,无需显式设置任何架构列.我们重用旧的DF模式,它是StructType类的,可以很容易地扩展.然而,这种方法有时是不可能的,并且在某些情况下可能比第一种方法效率低.

Answer 5

假设您有一个,DataFrame并且您希望通过将其转换为字段数据来对其进行一些修改RDD[Row].

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

要转换回DataFrame从RDD我们需要定义结构类型的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)

Answer 6

下面是将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|
+---+

Answer 7

方法1:(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()

方法2:(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.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