복붙노트

[SCALA] 어떻게 스파크에 dataframe에 RDD 개체를 변환하는

SCALA

어떻게 스파크에 dataframe에 RDD 개체를 변환하는

어떻게 Dataframe org.apache.spark.sql.DataFrame에 RDD (org.apache.spark.rdd.RDD [org.apache.spark.sql.Row])로 변환 할 수 있습니다. 나는 .rdd 사용 RDD하는 dataframe을 변환. 를 처리 한 후 내가 다시 dataframe에서 원하는. 이걸 어떻게 할 수 있습니까?

해결법

  1. ==============================

    1.는 SqlContext는 RDD 주어진 DataFrame을 만들 createDataFrame 방법을 가지고 있습니다. 나는 이들 중 하나가 당신의 상황에 대해 작동합니다 상상한다.

    는 SqlContext는 RDD 주어진 DataFrame을 만들 createDataFrame 방법을 가지고 있습니다. 나는 이들 중 하나가 당신의 상황에 대해 작동합니다 상상한다.

    예를 들면 :

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

    2.이 코드는 스칼라 2.11 스파크 2.X에서 완벽하게 작동

    이 코드는 스칼라 2.11 스파크 2.X에서 완벽하게 작동

    가져 오기 필요한 클래스

    import org.apache.spark.sql.{Row, SparkSession}
import org.apache.spark.sql.types.{DoubleType, StringType, StructField, StructType}

    여기가 스파크의, SparkSession 개체 만들기

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

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

    이 방법은 입력 RDD 형 RDD [행]이어야 필요하다.

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

    이제) (createDataFrame에 rowsRdd 및 스키마를 모두 적용

    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|
+------+----+----+
  3. ==============================

    3.당신의 RDD [행] 가정하면 RDD라고, 당신은 사용할 수 있습니다 :

    당신의 RDD [행] 가정하면 RDD라고, 당신은 사용할 수 있습니다 :

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

    4.참고 :이 대답은 원래는 여기에 게시 된

    참고 :이 대답은 원래는 여기에 게시 된

    나는 다른 답변에서 찾을 수 없습니다 사용할 수있는 옵션에 대한 추가 정보를 공유하고 싶습니다 때문에 나는이 대답을 게시하고

    행의 RDD에서 DataFrame을 만들려면 두 가지 옵션이 있습니다 :

    이미 지적한 바와 같이 1), 당신은 수입 sqlContext.implicits._에서 가져올 수 toDF ()를 사용할 수 있습니다. 그러나이 방법은 RDDs 다음과 같은 유형의 작동 :

    (출처 : SQLContext.implicits 객체 Scaladoc)

    마지막 서명은 실제로 (튜플 및 사례 클래스 scala.Product의 서브 클래스이기 때문에)은 튜플의 RDD 또는 케이스 클래스의 RDD 위해 일할 수 있다는 것을 의미한다.

    그래서, RDD [행]이 방법을 사용하기 위해, 당신은 RDD [: scala.Product T <]에 매핑해야합니다. 이는 다음 코드 조각에서와 같이 사용자 정의 케이스 클래스 또는 튜플로 각 행을 매핑하여 수행 할 수 있습니다 :

    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, 컬럼의 스키마를 설정해야한다는 것입니다. 당신이 사전에 스키마를 모르겠지만, 일이 좀 지저분를 얻을 수 있다면 아마이 프로그래밍 방식으로 수행 할 수 있습니다. 그래서, 대안으로, 또 다른 옵션이 있습니다 :

    는 SqlContext 개체에서 사용할 수있는 허용 대답에서와 같이 StructType) : RDD [행], 스키마 : 2) 당신은 createDataFrame (rowRDD를 사용할 수 있습니다. 오래된 DataFrame의 RDD를 변환하는 예 :

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

    명시 적으로 스키마 열을 설정할 필요가 없음을 유의하십시오. 우리는 StructType 클래스이며 쉽게 확장 할 수 있습니다 이전 DF의 스키마를 다시 사용합니다. 그러나이 방법은 때때로 불가능하고, 어떤 경우에는 첫 번째보다 효율적으로 할 수있다.

  5. ==============================

    5.당신이 DataFrame이 있고 RDD [행]로 변환하여 필드 데이터에 대한 몇 가지 수정을하고 싶은 가정하자.

    당신이 DataFrame이 있고 RDD [행]로 변환하여 필드 데이터에 대한 몇 가지 수정을하고 싶은 가정하자.

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

    RDD에서 DataFrame로 다시 변환하려면 우리는 RDD의 구조 유형을 정의 할 필요가있다.

    데이터 유형이 긴 있었다면 그것은 구조 LongType로 될 것입니다.

    구조 StringType은 다음 문자열합니다.

    val aStruct = new StructType(Array(StructField("id",LongType,nullable = true),StructField("role",StringType,nullable = true)))

    이제 createDataFrame 방법을 사용하여 DataFrame에 RDD을 변환 할 수 있습니다.

    val aNamedDF = sqlContext.createDataFrame(aRdd,aStruct)
  6. ==============================

    6.여기에 스파크 RDD으로 목록을 변환 한 후 Dataframe에 그 불꽃 RDD를 변환하는 간단한 예입니다.

    여기에 스파크 RDD으로 목록을 변환 한 후 Dataframe에 그 불꽃 RDD를 변환하는 간단한 예입니다.

    나는 다음과 같은 코드를 실행 REPL 스파크 쉘의 스칼라를 사용하고 있습니다, 여기에 SC는 스파크 쉘에서 암시로 볼 수 있습니다 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|
+---+
  7. ==============================

    7.방법 1 : (스칼라)

    방법 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()

    DF RDD로 변환하는 경우 클래스 적용된 후 로우 객체의 값을 추출하고,

    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
  8. ==============================

    8.스파크의 최신 버전 (2.0 이상)에

    스파크의 최신 버전 (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: _*)
  9. ==============================

    9.

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

    발 불꽃을 가정하는 것은 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 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
  10. ==============================

    10.DataFrame 또는 데이터 집합, 우아하게 다음 작품에 배열 [행] 변환하려면 :

    DataFrame 또는 데이터 집합, 우아하게 다음 작품에 배열 [행] 변환하려면 :

    말, 스키마는 행에 대한 StructType는,이다

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

    11. from https://stackoverflow.com/questions/29383578/how-to-convert-rdd-object-to-dataframe-in-spark by cc-by-sa and MIT license

'SCALA' 카테고리의 다른 글

[SCALA] HashPartitioner는 어떻게 작동합니까?  (0) 2019.10.29
[SCALA] 유형의 차이를 적용  (0) 2019.10.29
[SCALA] 스파크-CSV를 사용하여 하나의 CSV 파일을 쓰기  (0) 2019.10.29
[SCALA] 무엇 =의 차이> () =>, 그리고 단위 =>  (0) 2019.10.29
[SCALA] 케이스 스칼라에 열거 대 개체  (0) 2019.10.29
,

티스토리툴바