21. Hive 메타스토어 사용하기

Spark 데이터베이스와 테이블(1)

---

• 카탈로그 : 테이블과 뷰에 관한 메타 데이터 관리
  • 기본으로 메모리 기반 카탈로그 제공 - 세션이 끝나면 사라짐
  • Hive와 호환되는 카탈로그 제공 - Persistent
• 테이블 관리 방식
  • 데이터들은 데이터베이스라 부르는 폴더와 같은 구조로 관리(2단계)

 

Spark 데이터베이스와 테이블(2)

---

• 메모리 기반 테이블/뷰:
  • 임시 테이블로 앞서 사용해봤음
• 스토리지 기반 테이블
  • 기본적으로 HDFS와 Parquet 포맷을 사용
  • Hive와 호환되는 메타스토어 사용
  • 두 종류의 테이블이 존재(Hive와 동일한 개념)
    • Managed Table
      • Spark이 실제 데이터와 메타 데이터 모두 관리
      • Drop 테이블 하면 실제 HDFS에서 삭제
    • Unmanaged (External) Table
      • Spark이 메타 데이터만 관리
      • 실제 데이터 관리 X
      • Drop하면 실제 데이터 삭제 X

 

Spark SQL - Managed Table 사용 방법

---

• 두 가지 테이블 생성 방법
  • dataframe.saveAsTable("테이블이름")
  • SQL 문법 사용 (CREATE TABLE, CTAS)
• spark.sql.warehouse.dir가 가리키는 위치에 데이터가 저장됨
  • PARQUET이 기본 데이터 포맷
• 선호하는 테이블 타입
• Spark 테이블로 처리하는 것의 장점(파일로 저장하는 것 비교시)
  • JDBC/ODBC 등으로 Spark을 연결해서 접근 가능(태블로, 파워BI)

 

Spark SQL - External Table 사용 방법

---

• 이미 HDFS에 존재하는 데이터에 스키마를 정의해서 사용
  • LOCATION이란 프로퍼티 사용
• 메타데이터만 카탈로그에 기록됨
  • 데이터는 이미 존재.
  • External Table은 삭제되어도 데이터는 그대로임

CREATE TABLE table_name (
	column1 type1,
    column2 type2,
    column3 type3,
    ...
)
USING PARQUET
LOCATION 'hdfs_path';

 

Spark 데이터베이스/테이블 실습

---

• DataFrame을 Managed Table로 저장하기
• "default"말고 새로운 데이터베이스 사용해보기
• Spark SQL로 Managed Table 사용해보기(CTAS)

 

(1) Spark session 빌드

from pyspark.sql import SparkSession

# enableHiveSupport를 쓰면, Hive 메타스토어를 사용하는 세션으로 빌드된다.
spark = SparkSession \
    .builder \
    .appName("Python Spark Hive") \
    .enableHiveSupport() \
    .getOrCreate()

 

(2) Spark read csv

• inferSchema : 스키마 추론
• header : csv에 헤더 있음
• sep : 구분자는 \t

# Read in data and create a DataFrame
df = spark.read.csv("./data/orders.csv", inferSchema=True, header=True, sep ='\t')

 

(3) 메타스토어 내 TEST_DB라는 데이터베이스 생성

spark.sql("CREATE DATABASE IF NOT EXISTS TEST_DB")
spark.sql("USE TEST_DB")

 

(4) 데이터베이스 SHOW

spark.sql("SHOW DATABASES").show()

 

(5) TEST_DB 내에 orders라는 테이블 생성

# local standalone에서는 HDFS가 없기 때문에 로컬 디스크를 HDFS 처럼 사용한다.
df.write.saveAsTable("TEST_DB.orders", mode="overwrite")

 

(6) orders 테이블 확인

spark.sql("SELECT * FROM TEST_DB.orders").show(5)

 

(7) 카탈로그 확인

spark.catalog.listTables()

• table type = 'Managed' : 테이블의 형태는 Spark에서 managed 타입
• isTemporary = False : 임시 테이블 아님

 

(8) 테이블 삭제

spark.sql("""
    DROP TABLE IF EXISTS TEST_DB.orders_count;
""")