[스파크 완벽 가이드] 9장 - 데이터소스

데이터소스 API의 구조

읽기 API 구조

DataFrameReader.format(...).option("key", "value").schema(...).load()

 

쓰기 API 구조

DataFrameWriter.format(...).option(...).partitionBy(...).bucketBy(...).sortBy(...).save()

 

CSV 파일

csvFile = spark.read.format("csv")\
  .option("header", "true")\
  .option("mode", "FAILFAST")\ # 읽기 모드 - 형식에 맞지 않는 레코드를 만나면 즉시 종료
  .option("inferschema", "true")\
  .load("경로")

 

JSON 파일

스파크에서는 JSON 파일을 사용할 때 줄로 구분된 JSON을 기본적으로 사용한다.

spark.read.format("json")\
  .option("mode", "FAILFAST")\
  .option("inferschema", "true")\
  .load("경로")

 

Parquet 파일

Parquet는 다양한 스토리지 최적화 기술을 제공하는 오픈소스로 만들어진 컬럼 기반의 데이터 저장 방식이다.

저장소 공간을 절약할 수 있고, 전체 파일을 읽는 대신 개별 컬럼을 읽을 수 있으며, 컬럼 기반의 압축 기능을 제공한다.

그리고 복합 데이터 타입을 지원한다.

 

Parquet 파일은 읽기 연산 시 JSON이나 CSV보다 훨씬 효율적으로 동작한다.

 

# Parquet는 스키마가 파일 자체에 내장되어 있어서 inferSchema가 거의 필요 없다.

spark.read.format("parquet")\
  .load("경로")

 

Parquet 옵션

 

ORC 파일

ORC는 하둡 워크로드를 위해 설계된 self-describing이며, 데이터 타입을 인식할 수 있는 컬럼 기반의 파일 포맷이다.

대규모 스트리밍 읽기에 최적화되어 있고, 필요한 로우를 신속하게 찾아낼 수 있다.

 

Parquet와의 차이점은 ORC는 Hive에 최적화되어 있고, Parquet는 Spark에 최적화되어 있다.

 

spark.read.format("orc")\
  .load("경로")

 

SQL 데이터베이스

데이터베이스는 원시 파일 형태가 아니므로 고려해야 할 옵션이 많다.

ex) 데이터베이스의 인증 정보나 접속과 관련된 옵션, 네트워크 상태 등

 

데이터베이스의 데이터를 읽고 쓰기 위해서는 스파크 classpath에 데이터베이스의 JDBC 드라이버를 추가하고, 적절한 JDBC 드라이버 jar 파일을 제공해야 한다.

 

driver = "org.sqlite.JDBC"
path = "경로"
url = "jdbc:sqlite:" + path
tablename = "테이블 이름"

dbDataFrame = spark.read.format("jdbc")\
  .option("url", url)\
  .option("dbtable", tablename)\
  .option("driver", driver)\
  .option("user", "username")\
  .option("password", "password")\
  .load()

 

 

스파크는 데이터베이스의 테이블에서 스키마 정보를 읽어 테이블에 존재하는 컬럼의 데이터 타입을 스파크의 데이터 타입으로 변환한다.

 

쿼리 푸쉬다운

스파크는 DataFrame을 만들기 전에 데이터베이스 자체에서 데이터를 필터링하도록 만들 수 있다.

 

데이터베이스 병렬로 읽기

dbDataFrame = spark.read.format("jdbc")\
  .option("url", url)\
  .option("dbtable", tablename)\
  .option("driver", driver)\
  .option("user", "username")\
  .option("password", "password")\
  .option("numPartitions", 10)\
  .load()

numPartitions 옵션을 사용해 읽기 및 쓰기용 동시 작업 수를 제한할 수 있는 최대 파티션 수를 설정할 수 있다.

 

슬라이딩 윈도우 기반의 파티셔닝

url = "jdbc:mysql://localhost:3306/mydatabase"
properties = {
    "user": "root",
    "password": "password",
    "driver": "com.mysql.cj.jdbc.Driver"
}


table = "mytable"
partition_column = "id"  # 파티셔닝에 사용할 컬럼
lower_bound = 1          # 파티션 시작 값
upper_bound = 1000000    # 파티션 끝 값
num_partitions = 10      # 파티션 수 (병렬로 처리될 스레드 수)

# 병렬로 데이터를 읽어오기
df = spark.read.jdbc(
    url=url,
    table=table,
    properties=properties,
    column=partition_column,
    lowerBound=lower_bound,
    upperBound=upper_bound,
    numPartitions=num_partitions
)

 

고급 I/O 개념

병렬로 데이터 읽기

여러 익스큐터가 같은 파일을 동시에 읽을 수는 없지만, 여러 파일을 동시에 읽을 수는 있다.

다수의 파일이 존재하는 폴더를 읽을 때, 개별 파일은 DataFrame의 파티션이 된다.

 

병렬로 데이터 쓰기

파일이나 데이터 수는 데이터를 쓰는 시점에 DataFrame이 가진 파티션 수에 따라 달라질 수 있다.

기본적으로 파티션당 하나의 파일이 작성된다.

 

<파티셔닝>

파티셔닝은 어떤 데이터를 어디에 저장할 것인지 제어할 수 있는 기능이다.

csvFile.limit(10).write\
  .mode("overwright")\
  .partitionBy("DEST_COUNTRY_NAME")\
  .save("경로")

 

<버켓팅>

동일한 버킷 ID를 가진 데이터가 하나의 물리적 파티션에 모두 모여 있기 때문에 데이터를 읽을 때 셔플을 피할 수 있다.