如何将rdd对象转换为spark中的dataframe
如何将RDD（ org.apache.spark.rdd.RDD[org.apache.spark.sql.Row] ）转换为Dataframe org.apache.spark.sql.DataFrame 。 我使用.rdd将dataframe转换为rdd。 处理完之后，我希望它回到数据框中。 我怎样才能做到这一点 ？
SqlContext有一些createDataFrame方法，在给定RDD创build一个DataFrame 。 我想其中之一将为您的情况下工作。
例如：
def createDataFrame(rowRDD: RDD[Row], schema: StructType): DataFrame
使用给定模式从包含行的RDD创buildDataFrame。
假设你的RDD [行]被称为rdd，你可以使用：
val sqlContext = new SQLContext(sc) import sqlContext.implicits._ rdd.toDF()
这个代码完全适用于Scala 2.11的Spark 2.x
导入必要的课程
import org.apache.spark.sql.{Row, SparkSession} import org.apache.spark.sql.types.{DoubleType, StringType, StructField, StructType}
创buildSparkSession对象，这里是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（问题的实际答案）
这种方式要求inputrdd应该是RDD[Row]typesRDD[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) ) )
创build模式
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| +------+----+----+
假设你有一个DataFrame ，你想通过将其转换为RDD[Row]来对字段数据进行一些修改。
val aRdd = aDF.map(x=>Row(x.getAs[Long]("id"),x.getAs[List[String]]("role").head))
要从RDD转换回DataFrame ，我们需要定义RDD的结构types 。
如果数据types是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)
注意：这个答案最初是张贴在这里
我发布这个答案，因为我想分享有关其他答案中找不到的可用选项的其他详细信息
要从行的RDD创buildDataFrame，有两个主要选项：
1）正如已经指出的，你可以使用toDF() ，它可以通过import sqlContext.implicits._ 。 但是，此方法仅适用于以下types的RDD：
-
RDD[Int] -
RDD[Long] -
RDD[String] -
RDD[T <: scala.Product]
（来源： SQLContext.implicits对象的SQLContext.implicits ）
最后的签名实际上意味着它可以用于元组的RDD或者case类的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的模式。 也许这可以通过编程的方式来完成，如果你事先不知道这个模式，那么事情可能会变得有点混乱。 所以，另外还有另外一个select：
2）您可以像接受的答案中一样使用createDataFrame(rowRDD: RDD[Row], schema: StructType) ，它在SQLContext对象中可用。 转换旧DataFrame的RDD的示例：
val rdd = oldDF.rdd val newDF = oldDF.sqlContext.createDataFrame(rdd, oldDF.schema)
请注意，不需要显式设置任何模式列。 我们重用旧的DF模式，该模式是StructType类的，可以很容易地扩展。 但是，这种方法有时是不可能的，在某些情况下可能比第一个效率低。
方法1 :(斯卡拉）
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 :(斯卡拉）
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
下面是一个将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| +---+
有4种方法来创build数据框。
- DataFrame API
- 以编程方式指定模式
- 案例分类
- toDF（）方法更多信息： https ： //www.youtube.com/watch？v = JtJeEZeu1NU
简单的方法你可以使用toDF（）方法，在应用到DF（）之前，你必须做出结构化的格式，比只应用到DF（）方法。 我认为这个video将帮助你https://www.youtube.com/watch?v=nsbjzpbCJV4
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
相同的步骤，但较less的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
在新版本的火花（2.0 +）。 这也将工作，即使没有sqlcontext可用。
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: _*)
