SQL Operations

상위: Spark

요약

Spark에서 데이터를 조작하는 다양한 SQL 연산을 DSL 방식과 SQL 방식 두 가지로 수행할 수 있습니다. SELECT, FILTER, GROUP BY, JOIN 등의 기본 연산부터 문자열 처리, 결측치 처리까지 다양한 기능을 제공합니다.

SQL 쿼리 기본 문법

• 데이터 조회: SELECT, WHERE
• 정렬: ORDER BY
• 중복 제거: DISTINCT
• 집계: GROUP BY, HAVING, 집계 함수 (COUNT, AVG, SUM)
• 조인: JOIN

DataFrame 생성

DSL 코드 방식

from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, StringType, IntegerType
from pyspark.sql import Row

# SparkSession 생성
spark = SparkSession.builder.appName("ExampleApp").getOrCreate()

# 스키마 정의
schema = StructType([
    StructField("name", StringType(), True),
    StructField("age", IntegerType(), True)
])

# 데이터 전처리 및 DF 생성
parts = spark.sparkContext.parallelize([("Mine", "28"), ("Filip", "29"), ("Jonathan", "30")])
people = parts.map(lambda p: Row(name=p[0], age=int(p[1].strip())))
df = spark.createDataFrame(people, schema)

df.show()

SQL 방식

# DF을 SQL에서 사용할 수 있도록 TempView 등록
df.createOrReplaceTempView("people")

result = spark.sql("""
    SELECT name, age
    FROM people
""")
result.show()

파일로부터 DataFrame 생성

DSL 방식

people_df = spark.read.option("header", "false") \
                      .option("inferSchema", "true") \
                      .csv("people.txt") \
                      .toDF("name", "age")
people_df.show()

SQL 방식

people_df.createOrReplaceTempView("people")
result = spark.sql("""SELECT name, age FROM people""")
result.show()

다양한 데이터 소스 로딩

# JSON
df = spark.read.json("filename.json")
df = spark.read.load("filename.json", format="json")

# Parquet
df = spark.read.load("filename.parquet")

# TXT
df = spark.read.text("filename.txt")

SELECT

DSL 방식

df.select("column1", "column2").show()

SQL 방식

spark.sql("""
  SELECT column1, column2
  FROM my_table
""").show()

SELECT + Expression / Filter

DSL 방식

# column2 값에 +1 하여 새 컬럼 출력
df.select(
    col("column1"),
    (col("column2") + 1).alias("column2_plus1")
).show()

# column1 > 'A' 필터링
df.filter(col("column1") > "A").show()

SQL 방식

-- SQL 방식으로 동일한 작업
spark.sql("""
  SELECT column1, column2 + 1 AS column2_plus1
  FROM my_table
""").show()

spark.sql("""
  SELECT *
  FROM my_table
  WHERE column1 > 'A'
""").show()

WHEN / ISIN

DSL 방식

from pyspark.sql.functions import when

# column2 > 100 이면 1, 아니면 0 → flag 컬럼
df.select(
    "column1",
    when(df.column2 > 100, 1)
    .otherwise(0)
    .alias("flag")
).show()

# column1 값이 A 또는 B 인 행만
df.filter(df.column1.isin("A", "B")).show()

SQL 방식

spark.sql("""
  SELECT column1,
         CASE WHEN column2 > 100 THEN 1 ELSE 0 END AS flag
  FROM my_table
""").show()

spark.sql("""
  SELECT *
  FROM my_table
  WHERE column1 IN ('A', 'B')
""").show()

LIKE / STARTSWITH / ENDSWITH

DSL 방식

# column1 값이 'A'로 시작하는지 여부
df.select(
    col("column1"),
    col("column1").startswith("A").alias("starts_with_A")
).show()

# column2 값이 "00"으로 끝나는지 여부
df.select(
    col("column2"),
    col("column2").cast("string").endswith("00").alias("ends_with_00")
).show()

# column1이 "A"와 정확히 일치하는지
df.select(
    col("column1"),
    col("column1").like("A").alias("is_A")
).show()

SQL 방식

spark.sql("""
  SELECT column1, column1 LIKE 'A%' AS starts_with_A
  FROM my_table
""").show()

spark.sql("""
  SELECT column2,
         CAST(column2 AS STRING) LIKE '%00' AS ends_with_00
  FROM my_table
""").show()

spark.sql("""
  SELECT column1, column1 = 'A' AS is_A
  FROM my_table
""").show()

SUBSTR / BETWEEN

DSL 방식

# column1에서 2번째부터 3글자 추출 → name 컬럼
df.select(df.column1.substr(2, 3).alias("name")).show()

# column2가 50~150 사이인지 여부
df.select(
    col("column1"), col("column2"),
    col("column2").between(50, 150).alias("is_between_50_150")
).show()

SQL 방식

spark.sql("""
  SELECT SUBSTRING(column1, 2, 3) AS name
  FROM my_table
""").show()

spark.sql("""
  SELECT column1, column2,
         column2 BETWEEN 50 AND 150 AS is_between_50_150
  FROM my_table
""").show()

UPDATE / REMOVE

컬럼명 변경

# DSL 방식
df.withColumnRenamed("column1", "alphabet") \
  .withColumnRenamed("column2", "number") \
  .show()

# SQL 방식
spark.sql("""
    SELECT column1 AS alphabet, column2 AS number
    FROM my_table
""").show()

컬럼 삭제

# DSL 방식
df.drop("column1").show()

# SQL 방식
spark.sql("SELECT column2 FROM my_table").show()

GROUP BY / COUNT

DSL 방식

df.groupBy("column1").count().show()

SQL 방식

spark.sql("""
    SELECT column1, COUNT(*) as count
    FROM my_table
    GROUP BY column1
""").show()

FILTER

DSL 방식

df.filter(df["column2"] > 200).show()

SQL 방식

spark.sql("""
    SELECT *
    FROM my_table
    WHERE column2 > 200
""").show()

SORT / ORDER BY

DSL 방식

df.sort(df["column1"].desc()).show()

df.orderBy(["column1", "column2"], ascending=[True, False]).show()

SQL 방식

spark.sql("""
    SELECT *
    FROM my_table
    ORDER BY column1 DESC
""").show()

spark.sql("""
    SELECT *
    FROM my_table
    ORDER BY column1 ASC, column2 DESC
""").show()

Missing & Replacing Values

DSL 방식

# 결측치 채우기
df.na.fill({"column1": "Unknown", "column2": 0}).show()

# 결측치 제거
df.na.drop().show()

# 값 치환
df.na.replace({"A": "Alpha", "B": "Beta"}).show()

SQL 방식

# COALESCE로 결측치 처리
spark.sql("""
    SELECT
        COALESCE(column1, 'Unknown') AS column1,
        COALESCE(column2, 0) AS column2
    FROM my_table
""").show()

# NULL 제거
spark.sql("""
    SELECT *
    FROM my_table
    WHERE column1 IS NOT NULL AND column2 IS NOT NULL 
""").show()

# CASE로 값 치환
spark.sql("""
    SELECT CASE
        WHEN column1 = 'A' THEN 'Alpha'
        WHEN column1 = 'B' THEN 'Beta'
        ELSE column1
    END AS column1, column2
    FROM my_table
""").show()

집계 함수

기본 집계

from pyspark.sql.functions import count, sum, avg, max, min

df.groupBy("category") \
  .agg(
      count("*").alias("count"),
      sum("amount").alias("total"),
      avg("price").alias("avg_price"),
      max("quantity").alias("max_qty"),
      min("quantity").alias("min_qty")
  ).show()

SQL 방식

spark.sql("""
    SELECT 
        category,
        COUNT(*) as count,
        SUM(amount) as total,
        AVG(price) as avg_price,
        MAX(quantity) as max_qty,
        MIN(quantity) as min_qty
    FROM products
    GROUP BY category
""").show()

윈도우 함수

from pyspark.sql.window import Window
from pyspark.sql.functions import row_number, rank, dense_rank

windowSpec = Window.partitionBy("department").orderBy(col("salary").desc())

df.withColumn("rank", rank().over(windowSpec)) \
  .withColumn("dense_rank", dense_rank().over(windowSpec)) \
  .withColumn("row_number", row_number().over(windowSpec)) \
  .show()

JOIN 연산

# Inner Join
df1.join(df2, df1.id == df2.id, "inner").show()

# Left Join
df1.join(df2, "id", "left").show()

# Right Join
df1.join(df2, "id", "right").show()

# Full Outer Join
df1.join(df2, "id", "outer").show()

성능 최적화 팁

1. 필요한 컬럼만 선택

# Bad
df.filter(...).select("col1")

# Good
df.select("col1", "filter_col").filter(...)

2. 조건 필터링은 일찍

# Good: 데이터를 먼저 줄임
df.filter(col("date") > "2024-01-01").groupBy("category").count()

3. Broadcast Join

작은 테이블을 broadcast해 shuffle 회피 → 상세는 Spark SQL.