[Spark] Schema Evolution

Schema Evolution

• 데이터가 쌓이고 난 이후에 변경되었을때(이전 스키마 버전으로 데이터가 쌓여있고 스키마가 변경되었을때) 어떻게 데이터를 저장하는지에 대한 메커니즘 용어
• 아래 같은 경우 schema마다 컬럼이 다른데 이를 parquet파일에서 어떻게 처리하는지 알아본다.

 

1. SparkSession 생성

from pyspark.sql import *
from pyspark.sql.functions import *

if __name__ == "__main__":
    spark = SparkSession \
        .builder \
        .appName("Spark Schema Evolution Demo") \
        .master("local[3]") \
        .getOrCreate()

데이터 호출

parquet files.zip

0.13MB

df1 = spark.read. \
    parquet("schema1.parquet")
    
df1.printSchema()
df1.show()

df2 = spark.read. \
    parquet("schema2.parquet")
    
df2.printSchema()
df2.show()

df3 = spark.read. \
    parquet("schema3.parquet")
    
df3.printSchema()
df3.show()

 

위 3개의 parquet파일들을 한번에 불러올 수 있다.

• mergeSchema을 True로 세팅
• 3개의 파일을 하나의 데이터프레임으로 로딩된다.

df = spark.read. \
    option("mergeSchema", True). \
    parquet("*.parquet")
    
df.printSchema()
df.show(10)