시작하기

import org.apache.spark.sql.SparkSession

val spark = SparkSession
  .builder()
  .appName("Spark SQL basic example")
  .config("spark.some.config.option", "some-value")
  .getOrCreate()

// RDD를 DataFrame으로 바꾸는 것과 같은 암시적 변환(implicit conversion)을 처리하기 위해
// 아래와 같이 import를 해 줍니다.
import spark.implicits._

from pyspark.sql import SparkSession

spark = SparkSession \
    .builder \
    .appName("Python Spark SQL basic example") \
    .config("spark.some.config.option", "some-value") \
    .getOrCreate()

val df = spark.read.json("examples/src/main/resources/people.json")

// 표준출력(stdout)에 DataFrame의 내용을 보여줍니다.
df.show()
// +----+-------+
// | age|   name|
// +----+-------+
// |null|Michael|
// |  30|   Andy|
// |  19| Justin|
// +----+-------+

# 여기에서 spark는 이미 생성된 SparkSession입니다.
df = spark.read.json("examples/src/main/resources/people.json")
# 표준출력(stdout)에 DataFrame의 내용을 보여줍니다.
df.show()
# +----+-------+
# | age|   name|
# +----+-------+
# |null|Michael|
# |  30|   Andy|
# |  19| Justin|
# +----+-------+

// $-notation을 사용하기 위해 임포트합니다.
import spark.implicits._
// 트리 형태로 스키마를 출력합니다.
df.printSchema()
// root
// |-- age: long (nullable = true)
// |-- name: string (nullable = true)

// "name" 컬럼을 선택합니다.
df.select("name").show()
// +-------+
// |   name|
// +-------+
// |Michael|
// |   Andy|
// | Justin|
// +-------+

// 모든 사람을 선택하고, 나이를 1씩 증가시킵니다.
df.select($"name", $"age" + 1).show()
// +-------+---------+
// |   name|(age + 1)|
// +-------+---------+
// |Michael|     null|
// |   Andy|       31|
// | Justin|       20|
// +-------+---------+

// 나이가 21살보다 많은 사람을 선택합니다.
df.filter($"age" > 21).show()
// +---+----+
// |age|name|
// +---+----+
// | 30|Andy|
// +---+----+

// 각 나이별로 사람의 수를 셉니다.
df.groupBy("age").count().show()
// +----+-----+
// | age|count|
// +----+-----+
// |  19|    1|
// |null|    1|
// |  30|    1|
// +----+-----+

# spark와 df는 이전 예제와 동일합니다.
# 트리 형태로 스키마를 출력합니다.
df.printSchema()
# root
# |-- age: long (nullable = true)
# |-- name: string (nullable = true)

# "name" 컬럼을 선택합니다.
df.select("name").show()
# +-------+
# |   name|
# +-------+
# |Michael|
# |   Andy|
# | Justin|
# +-------+

# 모든 사람을 선택하고, 나이를 1씩 증가시킵니다.
df.select(df['name'], df['age'] + 1).show()
# +-------+---------+
# |   name|(age + 1)|
# +-------+---------+
# |Michael|     null|
# |   Andy|       31|
# | Justin|       20|
# +-------+---------+

# 나이가 21살보다 많은 사람을 선택합니다.
df.filter(df['age'] > 21).show()
# +---+----+
# |age|name|
# +---+----+
# | 30|Andy|
# +---+----+

# 각 나이별로 사람의 수를 셉니다.
df.groupBy("age").count().show()
# +----+-----+
# | age|count|
# +----+-----+
# |  19|    1|
# |null|    1|
# |  30|    1|
# +----+-----+

// DataFrame을 SQL 임시 뷰로 등록합니다.
df.createOrReplaceTempView("people")

val sqlDF = spark.sql("SELECT * FROM people")
sqlDF.show()
// +----+-------+
// | age|   name|
// +----+-------+
// |null|Michael|
// |  30|   Andy|
// |  19| Justin|
// +----+-------+

# DataFrame을 SQL 임시 뷰로 등록합니다.
df.createOrReplaceTempView("people")

sqlDF = spark.sql("SELECT * FROM people")
sqlDF.show()
# +----+-------+
# | age|   name|
# +----+-------+
# |null|Michael|
# |  30|   Andy|
# |  19| Justin|
# +----+-------+

// DataFrame을 전역 임시 뷰에 등록합니다.
df.createGlobalTempView("people")

// 전역 임시 뷰는 시스템 데이터베이스에 `global_temp` 라는 이름으로 등록됩니다.
spark.sql("SELECT * FROM global_temp.people").show()
// +----+-------+
// | age|   name|
// +----+-------+
// |null|Michael|
// |  30|   Andy|
// |  19| Justin|
// +----+-------+

// 전역 임시 뷰는 다른 SparkSession에서도 사용할 수 있습니다.
spark.newSession().sql("SELECT * FROM global_temp.people").show()
// +----+-------+
// | age|   name|
// +----+-------+
// |null|Michael|
// |  30|   Andy|
// |  19| Justin|
// +----+-------+

# DataFrame을 전역 임시 뷰에 등록합니다.
df.createGlobalTempView("people")

# 전역 임시 뷰는 시스템 데이터베이스에 `global_temp` 라는 이름으로 등록됩니다.
spark.sql("SELECT * FROM global_temp.people").show()
# +----+-------+
# | age|   name|
# +----+-------+
# |null|Michael|
# |  30|   Andy|
# |  19| Justin|
# +----+-------+

# 전역 임시 뷰는 다른 SparkSession에서도 사용할 수 있습니다.
spark.newSession().sql("SELECT * FROM global_temp.people").show()
# +----+-------+
# | age|   name|
# +----+-------+
# |null|Michael|
# |  30|   Andy|
# |  19| Justin|
# +----+-------+

case class Person(name: String, age: Long)

// Encoder는 케이스 클래스별로 생성됩니다.
val caseClassDS = Seq(Person("Andy", 32)).toDS()
caseClassDS.show()
// +----+---+
// |name|age|
// +----+---+
// |Andy| 32|
// +----+---+

// 일반적으로 사용되는 대부분의 타입에 대한 인코더는 spark.implicits._를 임포트하면 자동으로 포함됩니다.
val primitiveDS = Seq(1, 2, 3).toDS()
primitiveDS.map(_ + 1).collect() // Array(2, 3, 4) 반환.

// 클래스를 지정함으로써 DataFrame을 Dataset으로 변환할 수 있습니다. 값들은 클래스의 속성 변수 이름에 따라 자동으로 할당됩니다.
val path = "examples/src/main/resources/people.json"
val peopleDS = spark.read.json(path).as[Person]
peopleDS.show()
// +----+-------+
// | age|   name|
// +----+-------+
// |null|Michael|
// |  30|   Andy|
// |  19| Justin|
// +----+-------+

// RDD를 DataFrame으로 암시적으로 변환(implicit conversion)하기 위해 임포트합니다.
import spark.implicits._

// Create an RDD of Person objects from a text file, convert it to a Dataframe
// 텍스트 파일을 읽어 Person 객체의 RDD를 생성하고, 이를 DataFrame으로 변환합니다.
val peopleDF = spark.sparkContext
  .textFile("examples/src/main/resources/people.txt")
  .map(_.split(","))
  .map(attributes => Person(attributes(0), attributes(1).trim.toInt))
  .toDF()
// DataFrame을 임시 뷰로 등록합니다.
peopleDF.createOrReplaceTempView("people")

// 스파크에서 제공하는 sql 메소드를 이용해서 SQL문을 실행할 수 있습니다.
val teenagersDF = spark.sql("SELECT name, age FROM people WHERE age BETWEEN 13 AND 19")

// 결과에서 각 로우의 컬럼은 필드의 인덱스로 접근할 수 있습니다.
teenagersDF.map(teenager => "Name: " + teenager(0)).show()
// +------------+
// |       value|
// +------------+
// |Name: Justin|
// +------------+

// 필드의 이름으로 접근할 수도 있습니다.
teenagersDF.map(teenager => "Name: " + teenager.getAs[String]("name")).show()
// +------------+
// |       value|
// +------------+
// |Name: Justin|
// +------------+

// Dataset[Map[K,V]]에 대한 인코더가 선언되어 있지 않으므로, 여기서 명시적으로 선언해줍니다.
implicit val mapEncoder = org.apache.spark.sql.Encoders.kryo[Map[String, Any]]
// 기본 타입과 케이스 클래스는 아래와 같이 정의할 수 있습니다:
// implicit val stringIntMapEncoder: Encoder[Map[String, Any]] = ExpressionEncoder()

// row.getValuesMap[T]는 여러 개의 컬럼을 한번에 Map[String, T] 형태로 가져옵니다.
teenagersDF.map(teenager => teenager.getValuesMap[Any](List("name", "age"))).collect()
// Array(Map("name" -> "Justin", "age" -> 19))

from pyspark.sql import Row

sc = spark.sparkContext

# 텍스트 파일을 불러와 각 라인을 로우(row)로 변환합니다.
lines = sc.textFile("examples/src/main/resources/people.txt")
parts = lines.map(lambda l: l.split(","))
people = parts.map(lambda p: Row(name=p[0], age=int(p[1])))

# 스키마를 추론하고, DataFrame을 테이블로 등록합니다.
schemaPeople = spark.createDataFrame(people)
schemaPeople.createOrReplaceTempView("people")

# 테이블의 형태로 등록된 Dataframe에서 SQL 문을 실행할 수 있습니다.
teenagers = spark.sql("SELECT name FROM people WHERE age >= 13 AND age <= 19")

# SQL 쿼리의 결과는 DataFrame 객체가 됩니다.
# rdd는 :class:`Row`의 :class:`pyspark.RDD` 타입으로 내용을 반환합니다.
teenNames = teenagers.rdd.map(lambda p: "Name: " + p.name).collect()
for name in teenNames:
    print(name)
# Name: Justin

import org.apache.spark.sql.types._

// RDD를 생성합니다.
val peopleRDD = spark.sparkContext.textFile("examples/src/main/resources/people.txt")

// 스키마는 문자열로 인코딩됩니다.
val schemaString = "name age"

// 스키마 문자열을 기반으로 스키마를 생성합니다.
val fields = schemaString.split(" ")
  .map(fieldName => StructField(fieldName, StringType, nullable = true))
val schema = StructType(fields)

// RDD(people)에 들어 있는 레코드를 Row 객체로 변환합니다.
val rowRDD = peopleRDD
  .map(_.split(","))
  .map(attributes => Row(attributes(0), attributes(1).trim))

// RDD에 스키마를 적용합니다.
val peopleDF = spark.createDataFrame(rowRDD, schema)

// DataFrame을 사용하여 임시 뷰를 생성합니다.
peopleDF.createOrReplaceTempView("people")

// 테이블의 형태로 등록된 Dataframe에서 SQL 문을 실행할 수 있습니다.
val results = spark.sql("SELECT name FROM people")

// SQL 쿼리의 결과 역시 DataFrame이 되며, 모든 일반적인 RDD 동작을 지원합니다.
// 각 컬럼의 로우는 필드의 인덱스 값 또는 이름으로 접근할 수 있습니다.
results.map(attributes => "Name: " + attributes(0)).show()
// +-------------+
// |        value|
// +-------------+
// |Name: Michael|
// |   Name: Andy|
// | Name: Justin|
// +-------------+

# 데이터 타입을 임포트합니다.
from pyspark.sql.types import *

sc = spark.sparkContext

# 텍스트 파일을 불러와 각 라인을 로우(row)로 변환합니다.
lines = sc.textFile("examples/src/main/resources/people.txt")
parts = lines.map(lambda l: l.split(","))
# 각 라인을 튜플로 변환합니다.
people = parts.map(lambda p: (p[0], p[1].strip()))

# 스키마는 문자열로 인코딩됩니다.
schemaString = "name age"

fields = [StructField(field_name, StringType(), True) for field_name in schemaString.split()]
schema = StructType(fields)

# RDD에 스키마를 적용합니다.
schemaPeople = spark.createDataFrame(people, schema)

# DataFrame을 사용하여 임시 뷰를 생성합니다.
schemaPeople.createOrReplaceTempView("people")

# 테이블의 형태로 등록된 Dataframe에서 SQL 문을 실행할 수 있습니다.
results = spark.sql("SELECT name FROM people")

results.show()
# +-------+
# |   name|
# +-------+
# |Michael|
# |   Andy|
# | Justin|
# +-------+

import org.apache.spark.sql.{Row, SparkSession}
import org.apache.spark.sql.expressions.MutableAggregationBuffer
import org.apache.spark.sql.expressions.UserDefinedAggregateFunction
import org.apache.spark.sql.types._

object MyAverage extends UserDefinedAggregateFunction {
  // 이 집계 함수 입력 인자의 데이터 타입.
  def inputSchema: StructType = StructType(StructField("inputColumn", LongType) :: Nil)
  // 집계 버퍼 값의 데이터 타입.
  def bufferSchema: StructType = {
    StructType(StructField("sum", LongType) :: StructField("count", LongType) :: Nil)
  }
  // 반환 값의 데이터 타입.
  def dataType: DataType = DoubleType
  // 동일한 입력에 대해 항상 동일한 출력 값을 반환하는지(=결정적 연산(deterministic operation)) 여부.
  def deterministic: Boolean = true
  // Initializes the given aggregation buffer. The buffer itself is a `Row` that in addition to
  // standard methods like retrieving a value at an index (e.g., get(), getBoolean()), provides
  // the opportunity to update its values. Note that arrays and maps inside the buffer are still
  // immutable.
  // 주어진 집계 버퍼를 초기화합니다. 버퍼는 `Row` 객체이며 그 자체가 특정 인덱스에 해당하는 값을 반환하거나 값을 바꿀 수 있는 표준 메소드가 됩니다. (예: get(), getBoolean()) 버퍼 내의 배열과 맵은 값을 바꿀 수 없습니다.
  def initialize(buffer: MutableAggregationBuffer): Unit = {
    buffer(0) = 0L
    buffer(1) = 0L
  }
  // 새로운 입력 데이터 `input`을 받아 집계 버퍼 `buffer`를 업데이트합니다.
  def update(buffer: MutableAggregationBuffer, input: Row): Unit = {
    if (!input.isNullAt(0)) {
      buffer(0) = buffer.getLong(0) + input.getLong(0)
      buffer(1) = buffer.getLong(1) + 1
    }
  }
  // 두 집계 버퍼를 병합하고 그 값을 `buffer1`에 저장합니다.
  def merge(buffer1: MutableAggregationBuffer, buffer2: Row): Unit = {
    buffer1(0) = buffer1.getLong(0) + buffer2.getLong(0)
    buffer1(1) = buffer1.getLong(1) + buffer2.getLong(1)
  }
  // 최종 결과를 계산합니다.
  def evaluate(buffer: Row): Double = buffer.getLong(0).toDouble / buffer.getLong(1)
}

// 함수를 사용할 수 있도록 등록합니다.
spark.udf.register("myAverage", MyAverage)

val df = spark.read.json("examples/src/main/resources/employees.json")
df.createOrReplaceTempView("employees")
df.show()
// +-------+------+
// |   name|salary|
// +-------+------+
// |Michael|  3000|
// |   Andy|  4500|
// | Justin|  3500|
// |  Berta|  4000|
// +-------+------+

val result = spark.sql("SELECT myAverage(salary) as average_salary FROM employees")
result.show()
// +--------------+
// |average_salary|
// +--------------+
// |        3750.0|
// +--------------+

import org.apache.spark.sql.{Encoder, Encoders, SparkSession}
import org.apache.spark.sql.expressions.Aggregator

case class Employee(name: String, salary: Long)
case class Average(var sum: Long, var count: Long)

object MyAverage extends Aggregator[Employee, Average, Double] {
  // 이 집계 함수의 영값(zero value)입니다. 임의의 b 값에 대해서 b + zero = b를 만족합니다.
  def zero: Average = Average(0L, 0L)
  // 두 값을 가지고 새로운 값을 생성합니다. 성능을 위해 새로운 객체를 만드는 대신,
  // 함수에서 직접 `buffer`를 수정하여 반환할 수도 있습니다.
  def reduce(buffer: Average, employee: Employee): Average = {
    buffer.sum += employee.salary
    buffer.count += 1
    buffer
  }
  // 두 중간값을 병합합니다.
  def merge(b1: Average, b2: Average): Average = {
    b1.sum += b2.sum
    b1.count += b2.count
    b1
  }
  // reduce 호출 결과를 최종 리턴값으로 변환합니다.
  def finish(reduction: Average): Double = reduction.sum.toDouble / reduction.count
  // 중간값 타입에 대한 인코더를 명시합니다.
  def bufferEncoder: Encoder[Average] = Encoders.product
  // 최종 출력값 타입에 대한 인코더를 명시합니다.
  def outputEncoder: Encoder[Double] = Encoders.scalaDouble
}

val ds = spark.read.json("examples/src/main/resources/employees.json").as[Employee]
ds.show()
// +-------+------+
// |   name|salary|
// +-------+------+
// |Michael|  3000|
// |   Andy|  4500|
// | Justin|  3500|
// |  Berta|  4000|
// +-------+------+

// 함수를 `TypedColumn`으로 변환하고 이름을 지정합니다.
val averageSalary = MyAverage.toColumn.name("average_salary")
val result = ds.select(averageSalary)
result.show()
// +--------------+
// |average_salary|
// +--------------+
// |        3750.0|
// +--------------+