어떻게 내가 RDD을 (변환 할 수 있습니다 org.apache.spark.rdd.RDD[org.apache.spark.sql.Row]Dataframe에) org.apache.spark.sql.DataFrame. 를 사용하여 데이터 프레임을 rdd로 변환했습니다 .rdd. 그것을 처리 한 후 데이터 프레임에 다시 넣고 싶습니다. 어떻게해야합니까?
답변
SqlContext주어진 createDataFrame을 만드는 많은 메소드가 있습니다 . 나는 이것들 중 하나가 당신의 상황에 효과적이라고 생각합니다.DataFrameRDD
예를 들면 다음과 같습니다.
def createDataFrame(rowRDD: RDD[Row], schema: StructType): DataFrame지정된 스키마를 사용하여 행을 포함하는 RDD에서 DataFrame을 만듭니다.
답변
이 코드는 Scala 2.11의 Spark 2.x 에서 완벽하게 작동합니다.
필요한 수업 가져 오기
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 RDDsRDD그것을 만들어 보자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()답변
다른 답변에서 찾지 못한 사용 가능한 옵션에 대한 추가 정보를 공유하고 싶기 때문에이 답변을 게시하고 있습니다.
RDD 행에서 DataFrame을 만들려면 다음 두 가지 주요 옵션이 있습니다.
1) 이미 지적했듯이에 toDF()의해 가져올 수있는 것을 사용할 수 있습니다 import sqlContext.implicits._. 그러나이 방법은 다음 유형의 RDD에만 적용됩니다.
RDD[Int]RDD[Long]RDD[String]RDD[T <: scala.Product]
(출처 : Scaladoc of the SQLContext.implicitsobject)
마지막 서명은 실제로 튜플의 RDD 또는 케이스 클래스의 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")이 접근법의 주요 단점은 필자가 맵 함수에서 결과적으로 DataFrame의 스키마를 열 단위로 명시 적으로 설정해야한다는 것입니다. 스키마를 미리 모르는 경우 프로그래밍 방식으로 수행 할 수는 있지만 약간 혼란 스러울 수 있습니다. 따라서 다른 옵션이 있습니다.
2) SQLContext 객체 createDataFrame(rowRDD: RDD[Row], schema: StructType)에서 사용할 수있는 허용 된 답변과 같이 사용할 수 있습니다 . 이전 DataFrame의 RDD를 변환하는 예 :
val rdd = oldDF.rdd
val newDF = oldDF.sqlContext.createDataFrame(rdd, oldDF.schema)스키마 열을 명시 적으로 설정할 필요가 없습니다. 우리는 이전 DF의 스키마를 재사용합니다. StructType클래스 는 클래스이며 쉽게 확장 할 수 있습니다. 그러나이 방법은 때로는 불가능하며 경우에 따라 첫 번째 방법보다 효율성이 떨어질 수 있습니다.
답변
a가 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)답변
다음은 List를 Spark RDD로 변환 한 다음 해당 Spark RDD를 데이터 프레임으로 변환하는 간단한 예입니다.
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|
+---+답변
방법 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 : (파이썬)
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 : (파이썬)
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()행 객체에서 값을 추출한 다음 케이스 클래스를 적용하여 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