[Spark] 아파치 스파크(spark) DataFrame 구현하기

import org.apache.spark.sql.SparkSession   

val spark = SparkSession
    .builder()
    .appName("myAppName")
    .master("master")
    .getOrCreate()

val data = List(row1, row2, row3)

val result: DataFrame = spark.createDataFrame(data) 

// 또는
// scala의 경우 암묵적 형변환이 가능하다.   
import spark.implicits._
val result: DataFrame = data.toDF()

result.show() // 행 과 레코드 확인   
result.limit(10).show() // 출력 갯수 10개 제한

result.show(truncate=false) // row 길이가 길어도 짤리지 않고 모두 표기   

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.Row

val spark = SparkSession.builder()
  .master("local[1]")
  .appName("SparkByExamples.com")
  .config("spark.driver.bindAddress", "127.0.0.1")
  .getOrCreate()

// StructType과 StructField를 사용    
val sf1 = StructField("name", StringType, nullable = true)
val sf2 = StructField("age", IntegerType, nullable = true)
val sf3 = StructField("job", StringType, nullable = true)
val schema = StructType(List(sf1, sf2, sf3))

val r1 = Row("mike", 10, "student")
val r2 = Row("mark", 11, "teacher")
val r3 = Row("herry", 10, "teacher")

val rows = spark.sparkContext.parallelize(List(r1, r2, r3))
// org.apache.spark.rdd.RDD[org.apache.spark.sql.Row]

val df = spark.createDataFrame(rows, schema)
// org.apache.spark.sql.DataFrame   

df.printSchema() // 스키마 확인

val df = spark.read.json("/people.json")  

// parquet 파일을 직접 FROM 절을 통해 바로 읽어 올 수도 있다.   
val df = spark.sql("SELECT * FROM parquet.`/users.parquet`")

// Write Programmatic Queries against DataFrame  
// $는 컬럼을 나타낸다.   
df.select($"name", $"age").filter($"age">21).groupBy("age").count()  

// Apply Functional Transformations to DataSet
df.map(row => row.getAs[Long]("age") + 10)

df.createOrReplaceTempView("people")

val sqlDf = spark.sql("SELECT * FROM people") 

// 또는   
val sqlDf = spark.sql(""" SELECT * FROM json.`/people.json` WHERE name LIKE "A%" """)   

val df=spark.read
    .format("avro")
    .load("/users.avro")

df.dtypes

df.printSchema

df.explain(true)

val df = Seq(("Nam", List("A", "B", "C", "D"))).toDF("name", "grade")

df.show()
+----+------------+
|name|       grade|
+----+------------+
| Nam|[A, B, C, D]|
+----+------------+

df.select(explode(col("grade")).alias("score")).show()   

+-----+
|score|
+-----+
|    A|
|    B|
|    C|
|    D|
+-----+

df.join(df2, "id").show // inner join


// 조인 조건에 해당하는 컬럼 하나 이상일 때 Array를 사용한다.
// ex) Array("age", "name")
df.join(df2, Array("age"), "left_outer").show // left_outer 조인 

df.join(df3, $"age" === $"age3").show

ticketDf.groupBy("ticketId").max("occurredAt")

// OR   
import org.apache.spark.sql.functions._
df.groupBy("age").agg(max("occurredAt"))

spark.udf.register("myUDF", (arg1: Int, arg2: String) => arg2 + arg1)    

import org.apache.spark.sql.functions._

val rdd: RDD[Row] = peopleDF.rdd
// org.apache.spark.rdd.RDD[org.apache.spark.sql.Row]    

peopleDF.rdd.collect
Array[org.apache.spark.sql.Row] = Array([null, Michael], [30, Andy], [19, Justin])

import spark.implicits._

val ticketDf = List(
      Ticket(1L, "closed"),
      Ticket(2L, "closed"),
      Ticket(3L, "open")
    ).toDF()

ticketRdd.foreach(row => {
    println("row(n) returns element in the nth column: " + row(1))

    println("row.fieldIndex returns index of the age column: " + row.fieldIndex("status"))

    println("row.getAs get correctly typed values: " + row.getAs[String]("status"))
})

// row(n) returns element in the nth column
// row.fieldIndex("status") returns index of the age column
// row.getAs[String]("status") get correctly typed values

// Use type-specific get methods to return typed values   
// - row.getString(n) returns nth column as a string
// - row.getInt(n) returns nth column as an integer    

import org.apache.spark.sql.functions.{col,when, count}
    def countCols(columns:Array[String]):Array[Column]={
      columns.map(c=>{
        count(when(col(c).isNull,c)).alias(c)
      })
    }
    df.select(countCols(df.columns):_*).show()