spark managed tables - partitionBy, bucketBy, sortBy

문제

• spark 에서 schema 를 적용해 데이터를 읽기에서 사용된 csv 형식의 데이터를 spark managed table 로 저장하기
• partitionBy, bucketBy, sortBy 도 사용해 보도록 한다.

 Code

/**
 * build.sbt 에 아래 의존성을 추가해 준다.
 * "org.apache.spark" %% "spark-hive" % sparkVersion
 */
import org.apache.spark.sql.{DataFrame, Encoders, SaveMode, SparkSession}

import java.sql.Date

final case class Person(name: String, age: Int, birthDay: Date)

object MySpark extends Serializable {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession
      .builder()
      .master("local[*]")
      .enableHiveSupport()
      .getOrCreate()

    val schemaDef = Encoders.product[Person].schema

    val df: DataFrame = spark.read
      .option("header", "true")
      .option("mode", "FAILFAST")
      .schema(schemaDef)
      .option("dateFormat", "yyyy-MM-dd")
      .csv("hdfs://localhost:9000/csv.txt")

    spark.sql("CREATE DATABASE IF NOT EXISTS MyDB")
//    spark.catalog.setCurrentDatabase("MyDB")
//    spark.catalog.listTables("MyDB").show()

    df.write
      .format("csv")
      .mode(SaveMode.Overwrite)
      //      .partitionBy("name")
      .bucketBy(2, "name")
      .sortBy("date")
      .saveAsTable("MyDB.my_tbl")
    spark.sql("select * from MyDB.my_tbl").show()

    spark.stop
  }
}

 

설명

• metastore 를 사용하기 위해 SparkSession instance 를 만들때, enableHiveSupport() 를 추가했다.
• local mode 로 spark 이 실행되었다면 metastore_db, spark-warehouse 디렉토리를 확인할 수 있다.
• bucketBy(2, "name") 에 의해 spark-warehouse 하위에 2개의 파일로 데이터가 저장된 것을 볼 수 있다.