Как преобразовать RDD ( org.apache.spark.rdd.RDD[org.apache.spark.sql.Row]
) в Dataframe org.apache.spark.sql.DataFrame
. Я преобразовал фрейм данных в rdd, используя .rdd
. После обработки я хочу вернуть его в фрейм данных. Как я могу это сделать ?
scala
apache-spark
apache-spark-sql
rdd
user568109
источник
источник
Ответы:
SparkSession
имеет рядcreateDataFrame
методов, которые создаютDataFrame
заданныйRDD
. Я полагаю, что один из них будет работать в вашем контексте.Например:
def createDataFrame(rowRDD: RDD[Row], schema: StructType): DataFrame
источник
Этот код отлично работает в 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| +------+----+----+
источник
Предполагая, что ваш RDD [row] называется rdd, вы можете использовать:
val sqlContext = new SQLContext(sc) import sqlContext.implicits._ rdd.toDF()
источник
Примечание: этот ответ был изначально размещен здесь
Я публикую этот ответ, потому что хотел бы поделиться дополнительной информацией о доступных вариантах, которые я не нашел в других ответах
Чтобы создать DataFrame из RDD строк, есть два основных варианта:
1) Как уже указывалось, вы можете использовать файл,
toDF()
который можно импортировать с помощьюimport sqlContext.implicits._
. Однако этот подход работает только для следующих типов RDD:RDD[Int]
RDD[Long]
RDD[String]
RDD[T <: scala.Product]
(источник: Scaladoc от
SQLContext.implicits
объекта)Последняя сигнатура фактически означает, что она может работать для RDD кортежей или RDD классов case (поскольку кортежи и классы case являются подклассами
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")
Главный недостаток этого подхода (на мой взгляд) заключается в том, что вам нужно явно задать схему результирующего DataFrame в функции карты, столбец за столбцом. Возможно, это можно сделать программно, если вы заранее не знаете схему, но там все может получиться немного запутанным. Итак, в качестве альтернативы есть еще один вариант:
2) Вы можете использовать
createDataFrame(rowRDD: RDD[Row], schema: StructType)
как в принятом ответе, который доступен в объекте SQLContext . Пример преобразования RDD старого DataFrame:val rdd = oldDF.rdd val newDF = oldDF.sqlContext.createDataFrame(rdd, oldDF.schema)
Обратите внимание, что нет необходимости явно устанавливать какой-либо столбец схемы. Мы повторно используем старую схему DF, которая является
StructType
классной и может быть легко расширена. Однако этот подход иногда невозможен, а в некоторых случаях может быть менее эффективным, чем первый.источник
import sqlContext.implicits.
Предположим, у вас есть
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)))
Теперь вы можете преобразовать RDD в DataFrame с помощью метода createDataFrame .
val aNamedDF = sqlContext.createDataFrame(aRdd,aStruct)
источник
Вот простой пример преобразования вашего списка в Spark RDD, а затем преобразования этого Spark RDD в Dataframe.
Обратите внимание, что я использовал scala REPL оболочки Spark для выполнения следующего кода. Здесь sc - это экземпляр SparkContext, который неявно доступен в оболочке Spark. Надеюсь, он ответит на ваш вопрос.
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| +---+
источник
Метод 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
источник
В более новых версиях 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: _*)
источник
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
источник
Я попытался объяснить решение, используя проблему подсчета слов . 1. Прочтите файл с помощью sc
Способы создания DF
Прочитать файл с помощью искры
val rdd=sc.textFile("D://cca175/data/")
Rdd к Dataframe
val df = sc.textFile ("D: // cca175 / data /") .toDF ("t1") df.show
Способ 1
Создать RDD для подсчета слов в Dataframe
val df=rdd.flatMap(x=>x.split(" ")).map(x=>(x,1)).reduceByKey((x,y)=>(x+y)).toDF("word","count")
Метод 2
Создать Dataframe из 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 = новый 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)))
Создать DataFrame из RDD со схемой
val df = spark.createDataFrame (rowRdd, schema)
df.show
источник
Чтобы преобразовать массив [Row] в DataFrame или Dataset, элегантно работает следующее:
Скажем, schema - это StructType для строки, тогда
val rows: Array[Row]=... implicit val encoder = RowEncoder.apply(schema) import spark.implicits._ rows.toDS
источник