如何在火花转换 rdd 对象的数据帧

小开

最佳答案

SparkSession 有许多 createDataFrame方法，它们在给定 RDD的情况下创建 DataFrame。我想其中一个会对你的背景有帮助。

例如:

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

方法从包含行的 RDD 创建 DataFrame 模式。

小开

假设您的 RDD [ row ]被称为 RDD，您可以使用:

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

小开

假设您有一个 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)

小开

下面是一个将 List 转换为 Spark RDD 然后将 Spark RDD 转换为 Dataframe 的简单示例。

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

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

小开

注: 此答案最初发布于给你

我发布这个答案是因为我想分享更多关于可用选项的细节，我没有在其他的答案 中找到

要从 RDD 的行创建 DataFrame，有两个主要选项:

1) 正如已经指出的，你可以使用 toDF()，它可以被 import sqlContext.implicits._导入。然而，这种方法只适用于下列类型的排放物:

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

(来源: SQLContext.implicits对象的斯卡拉多克)

最后一个签名实际上意味着它可以为元组的 RDD 或大小写类的 RDD 工作(因为元组和大小写类是 scala.Product的子类)。

因此，要对 RDD[Row]使用这种方法，必须将它映射到 RDD[T <: scala.Product]。这可以通过将每一行映射到自定义 case 类或元组来实现，如下面的代码片段所示:

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对象中可用。转换旧数据框架的 RDD 的示例:

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

注意，不需要显式设置任何模式列。我们重用了旧的 DF 模式，它是 StructType类的，可以很容易地扩展。然而，这种方法有时是不可能的，而且在某些情况下可能不如第一种方法有效。

小开

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

从 row 对象中提取值，然后应用 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

小开

One needs to create a schema, and attach it to the Rdd.

假设瓦尔・斯帕克是火花会议制造商的产品。

    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

小开

这段代码在 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))
)
)

# # 方法一使用 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|
+------+----+----+

小开

关于更新的版本的火花(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: _*)

小开

要将 Array [ Row ]转换为 DataFrame 或 Dataset，可以优雅地执行以下操作:

那么，模式就是行的 structType

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

小开

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

生成单词计数
创建 DF 的方法
- ToDF 方法
- ToDF (“ word”，“ count”)
  - CreateDataFrame (rdd，schema)
使用火花读取文件
```
val rdd=sc.textFile("D://cca175/data/")
```
Rdd 呼叫数据帧

Val df = sc.textFile (“ D://cca175/data/”) . toDF (“ t1”) DF 秀

方法1

创建字数 RDD 到数据帧
```
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

定义 Schema

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

Val schema = new structType (). Add (structField (“ word”，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 创建数据框架

Val df = spot.createDataFrame (rowRdd，schema)
Show

小开

我遇到了同样的问题，最后我解决了它。它很简单，很容易。

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

如何在火花转换 rdd 对象的数据帧

使用火花读取文件

Rdd 呼叫数据帧

方法1

创建字数 RDD 到数据帧

方法2

从 Rdd 创建数据框架

方法3

定义 Schema

创建 RowRDD

使用模式从 RDD 创建数据框架