Parquet 파일

// 일반적인 타입에 대한 인코더는 spark.implicits._ 를 import 함으로써 사용할 수 있습니다.
import spark.implicits._

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

// DataFrame은 스키마 정보를 유지하면서 Parquet 파일로 저장할 수 있습니다.
peopleDF.write.parquet("people.parquet")

// 위에서 생성한 parquet 파일을 읽습니다.
// parquet 파일에는 스키마 정보가 포함되어 있습니다. 따라서 위에서 읽어온 스키마는 그대로 보존됩니다.
// parquet 파일을 불러온 결과 역시 DataFrame이 됩니다.
val parquetFileDF = spark.read.parquet("people.parquet")

// parquet 파일은 임시 뷰를 생성하고 SQL문을 실행하는데 사용할 수 있습니다.
parquetFileDF.createOrReplaceTempView("parquetFile")
val namesDF = spark.sql("SELECT name FROM parquetFile WHERE age BETWEEN 13 AND 19")
namesDF.map(attributes => "Name: " + attributes(0)).show()
// +------------+
// |       value|
// +------------+
// |Name: Justin|
// +------------+

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

# DataFrame은 스키마 정보를 유지하면서 Parquet 파일로 저장할 수 있습니다.
peopleDF.write.parquet("people.parquet")

# 위에서 생성한 parquet 파일을 읽습니다.
# parquet 파일에는 스키마 정보가 포함되어 있습니다. 따라서 위에서 읽어온 스키마는 그대로 보존됩니다.
# parquet 파일을 불러온 결과 역시 DataFrame이 됩니다.
parquetFile = spark.read.parquet("people.parquet")

# parquet 파일은 임시 뷰를 생성하고 SQL문을 실행하는데 사용할 수 있습니다.
parquetFile.createOrReplaceTempView("parquetFile")
teenagers = spark.sql("SELECT name FROM parquetFile WHERE age >= 13 AND age <= 19")
teenagers.show()
# +------+
# |  name|
# +------+
# |Justin|
# +------+

// RDD를 DataFrame으로 변환하기 위해 사용됩니다.
import spark.implicits._

// 간단한 DataFrame을 생성하고 파티션 디렉토리에 저장합니다.
val squaresDF = spark.sparkContext.makeRDD(1 to 5).map(i => (i, i * i)).toDF("value", "square")
squaresDF.write.parquet("data/test_table/key=1")

// 새로운 컬럼을 추가하고 기존의 컬럼을 삭제함으로써 또 다른 DataFrame을 생성하고,
// 새로운 분할 디렉토리에 저장합니다.
val cubesDF = spark.sparkContext.makeRDD(6 to 10).map(i => (i, i * i * i)).toDF("value", "cube")
cubesDF.write.parquet("data/test_table/key=2")

// 분할된 테이블을 읽어옵니다.
val mergedDF = spark.read.option("mergeSchema", "true").parquet("data/test_table")
mergedDF.printSchema()

// The final schema consists of all 3 columns in the Parquet files together
// with the partitioning column appeared in the partition directory paths
// 최종 스키마는 Parquet 파일에 들어 있는 세 개의 컬럼과
// 파티션 디렉토리 경로에 나타나는 파티션 컬럼, 모두 네 개의 컬럼으로 이루어집니다.
// root
//  |-- value: int (nullable = true)
//  |-- square: int (nullable = true)
//  |-- cube: int (nullable = true)
//  |-- key: int (nullable = true)

from pyspark.sql import Row

# spark는 이전 예제에서 나온 것입니다.
# 간단한 DataFrame을 생성하고 분할 디렉토리에 저장합니다.
sc = spark.sparkContext

squaresDF = spark.createDataFrame(sc.parallelize(range(1, 6))
                                  .map(lambda i: Row(single=i, double=i ** 2)))
squaresDF.write.parquet("data/test_table/key=1")

# 새로운 컬럼을 추가하고 기존의 컬럼을 삭제함으로써 또 다른 DataFrame을 생성하고,
# 새로운 분할 디렉토리에 저장합니다.
cubesDF = spark.createDataFrame(sc.parallelize(range(6, 11))
                                .map(lambda i: Row(single=i, triple=i ** 3)))
cubesDF.write.parquet("data/test_table/key=2")

# 분할된 테이블을 읽어옵니다.
mergedDF = spark.read.option("mergeSchema", "true").parquet("data/test_table")
mergedDF.printSchema()

# 최종 스키마는 Parquet 파일에 들어 있는 세 개의 컬럼과
# 파티션 디렉토리 경로에 나타나는 파티션 컬럼, 모두 네 개의 컬럼으로 이루어집니다.
# root
#  |-- double: long (nullable = true)
#  |-- single: long (nullable = true)
#  |-- triple: long (nullable = true)
#  |-- key: integer (nullable = true)