[PySPARK] 스파크 SQL과 데이터 프레임 Part 1

Spark .. 엔지니어링 공부 때문에 약간 급하게 벼락치기 아닌 벼락치기 하면서 공부중 ^0^....아무렴 어때 화이팅

SparkSql 특징

• 다양한 구조화된 형식(예: JSON, Hive 테이블, Parquet, Avro, ORC, CSV)으로 데이터를 읽고 쓸 수 있다.
• Tableau, Power BI, Talend와 같은 외부 비즈니스 인텔리전스(BI)부터 MySQL 및 PostgreSQL과 같은 RDBMS등 에서 JDBC/ODBC 커넥터를 사용하여 데이터를 쿼리할 수 있다.
• Spark 애플리케이션의 데이터베이스에 테이블 또는 view로 저장된 정형 데이터와 상호 작용할 수 있는 프로그래밍 방식 인터페이스를 제공한다.
• 정형 데이터에 대해 SQL 쿼리를 실행하기 위한 대화형 셸을 제공한다.

SparkSql 사용하기

SparkSession은 정형화 API로 스파크를 프로그래밍하기 위한 진입점을 제공한다. SparkSession을 이용하면 쉽게 클래스를 가져오고 코드에서 인스턴스 생성 가능하다.

먼저 SQL 쿼리를 실행하기 위해서 spark라고 선언된 SparkSession 인스턴스에 sql() 함수를 사용한다. 이러한 형식의 모든 spark.sql() 쿼리는 추가적인 스파크 작업이 가능하도록 데이터 프레임을 반환한다.

spark.sql 예제

날짜, 지연거리, 출발지, 목적지 등 미국 항공편에 대한 데이터세트로 기본 쿼리 예제를 살펴보자.

데이터셋 - https://github.com/databricks/LearningSparkV2/tree/master/databricks-datasets/learning-spark-v2/flights

먼저 데이터세트를 임시 뷰로 읽어 SQL 쿼리를 사용할 준비를 한다.

from pyspark.sql import SparkSession
# create SparkSession named spark
spark = (SparkSession
        .builder
        .appName("SparkSQLExampleApp")
        .getOrCreate())

# dataset path
csv_file = "./data/departuredelays.csv"

# Read data and create temporary view
df = (spark.read.format("csv")
    .option("inferSchema", "true")
    .option("header", "true")
    .load(csv_file))
df.createOrReplaceTempView("us_delay_filghts_tbl")
df.show()

# 스키마 지정
# schema = "'data' STRING, 'delay' INT, 'origin' STRING, 'destination' STRING"

출력결과는 MAC 기준입니다 :)

spark.sql 예제 쿼리 실행

• <미국 항공편 운항 지연 데이트세트> 내 칼럼 정리
  • date(날짜): 02-19 09:25 AM으로 매핑되는 02190925와 같은 문자열 포함
  • delay(지연): 계획된 도착시간과 실제 도착시간의 차이를 분으로 제공 (조기도착: 음수)
  • distance(비행거리): 마일 단위로 출발지와 도착지의 거리 제공
  • origin(출발지): 출발지의 IATA 공항 코드
  • destination(도착지): 도착지의 IATA 공항 코드
• ex) 비행거리가 1,000마일 이상인 모든 항공편 조회하기

spark.sql("""SELECT distance, origin, destination
FROM us_delay_flights_tbl WHERE distance > 1000
ORDER BY distance DESC""").show(10)

 

ex.2 ) 샌프란시스코(SFO)와 시카고(ORD) 간 2시간 이상 지연이 있었던 모든 항공편

spark.sql("""SELECT date, delay, origin, destination
FROM us_delay_flights_tbl
WHERE delay > 120 AND ORIGIN = 'SFO' AND DESTINATION = 'ORD'
ORDER BY delay DESC""").show(10)

ex.3) 모든 미국 항공편에 대해 지연에 대한 표시를 레이블로 지정

df.select("date", "delay", "destination",
          when(df.delay > 360, "Very Long Delays")
          .when((df.delay >= 120) & (df.delay <= 360), "Long Delays")
          .when((df.delay >= 60) & (df.delay < 120), "Short Delays")
          .when((df.delay > 0) & (df.delay < 60), "Tolerable Delays")
          .when(df.delay == 0, "No Delays")
          .otherwise("Early").alias("Flight_Delays")
          ).orderBy("origin", df.delay.desc()).show(10)

SQL 테이블과 뷰

• 스파크는 각 테이블과 해당 데이터에 관련된 정보인 스키마, 설명, 테이블명, 데이터베이스명, 칼럼명, 파티션, 실제 데이터의 물리적 위치 등의 메타데이터를 가지고 있다. 이 모든 정보는 중앙 메타스토어에 저장된다.
• 스파크는 스파크 테이블만을 위한 별도 메타스토어를 생성하지 않고 기본적으로는 /user/hive/warehouse에 있는 아파치 하이브 메타스토어를 사용해 테이블에 대한 모든 메타데이터를 유지한다.

관리형 테이블과 비관리형 테이블

관리형 테이블 비관리형 테이블

관리범위	메타데이터와 파일 저장소의 데이터 모두 관리	오직 메타데이터만 관리
저장소	로컬 파일 시스템, HDFS, Amazon S3, Azure Blob 같은 객체 저장소	카산드라와 같은 외부 데이터 소스에서 데이터를 직접 관리
사용	DROP TABLE <테이블명>과 같은 SQL 명령은 메타데이터와 실제 데이터를 모두 삭제	동일 명령이 실제 데이터는 그대로 두고 메타데이터만 삭제

SQL 데이터베이스와 테이블 생성하기

• 테이블은 데이터베이스 안에 존재한다. 기본적으로 스파크는 dafault 데이터베이스 안에 테이블을 생성.
• 미국 항공 지연 데이터 사용해 관리형 및 비관리형 테이블 모두 생성해본다.

1. learn_spark_db라는 데이터베이스 생성 및 사용 선언하기

spark.sql("CREATE DATABASE learn_spark_db")
spark.sql("USE learn_spark_db")

2. 관리형 테이블 생성

spark.sql("CREATE TABLE managed_us_delay_flights_tbl (data STRING, delay INT, distance INT, origin STRING, destination STRING)")

#learn_spark_db 데이터베이스 안에 us_delay_flights라는 관리형 테이블 생성
csv_file = "/databricks-datasets/learning-spark-v2/flights/departuredelays.csv"

schema = "data STRING, delay INT, distance INT, origin STRING, destination STRING"
flights_df = spark.read.csv(csv_file, schema=schema)
flights_df.write.saveAsTable("managed_us_delay_flights_tbl")

3. 비관리형 테이블 생성

spark.sql("""
	CREATE TABLE us_delay_flights_tbl(
		date STRING,
		delay INT,
		distance INT,
		origin STRING,
		destination STRING)

USING csv OPTIONS (PATH '/databricks-datasets/learning-spark-v2/flights/departuredelays.csv')
""")
# 데이터 프레임 API에서 사용해야하는 명령어
(flights_df
	.write
	.option("path", "/tmp/data/us_flights_delay")
	.saveAsTable("us_delay_flights_tbl"))

뷰

뷰는 전역(해당 클러스터의 모든 SparkSession)또는 세션 범위(단일 SparkSession)일 수 있으며, 테이블과 달리 데이터를 소유하지 않기 때문에 일시적으로 스파크애플리케이션이 종료되면 사라짐.

임시뷰 예제

미국 비행 지연 데이터에서 뉴욕이 출발지인 공항이 있는 하위 데이터셋에 대해서만 작업하는 경우

CREATE OR REPLACE GLOBAL TEMP VIEW us_origin_airport_SFO_global_tmp_view AS
SELECT data,delay,origin,destination FROM us_delay_flights_tbl
WHERE origin=’JFK’;

df_jfk=spark.sql("select data, delay, origin, destination from us_delay_flights_tbl where origin='JFK'")

#create a temporary and global temporary view

df_jfk.createOrReplaceTempView("us_origin_airport_JFK_tmp_view")

이러한 뷰를 생성 한 후에는 테이블에 대해 수행하는 것처럼 쿼리를 실행할 수 있다.

스파크는 global_temp 라는 전역 임시 데이터베이스에 전역 임시 뷰를 생성하므로 해당 뷰에 액세스 할 때는 global_temp.<view_name>접두사를 이용해야 한다.

SELECT * FROM global_temp.us.origin_airport_JFK_global_tmp_view

반면 임시뷰는 global_temp접두사 없이 접근 가능하다.

SELECT * FROM us.origin_airport_JFK_tmp_view #SQL

spark.sql("SELECT * FROM us_origin_airport_JFK_temp_view") #Python

뷰 드롭

#SQL
DROP VIEW IF EXIST us.origin_airport_JFK_global_tmp_view

#Python
spark.catalog.dropGlobalTempview("us.origin_airport_JFK_global_tmp_view")
spark.catalog.dropTempview("us.origin_airport_JFK_tmp_view")

임시뷰는 스파크 애플리케이션 내의 단일 SparkSession에 연결된다.

전역 임시뷰는 스파크 애플리케이션 내의 여러 SparkSession에서 볼 수 있다.

---

메타데이터

스파크는 메타데이터를 스파크 SQL의 상위 추상화 모듈인 카탈로그에 저장한다.

#  // In Scala/Python
spark.catalog.listDatabases()
spark.catalog.listTables()
spark.catalog.listColumns("us_delay_flights_tbl")

테이블을 데이터 프레임으로 읽기

스파크 SQL 데이터베이스 및 테이블을 정리된 데이터로 로드

• 애플리케이션의 다운스트림에서 사용할 수 있음

기존 데이터 베이스 learn_spark_db와 테이블 us_delay_flights_tbl이 있을때,

외부 JSON 파일에서 읽는 대신 SQL을 사용하여 테이블을 쿼리하고 반환된 결과를 데이터프레임에 저장

us_flights_df = spark.sql(“SELECT * FROM us_delay_flights_tbl”)
us_flights_df2 = spark.table(“us_delay_flights_tbl”)

데이터 프레임 및 SQL 테이블을 위한 데이터 소스

스파크 SQL은

• 다양한 데이터 소스에 대한 인터페이스 제공
• 데이터 소스 API를 사용하여 이러한 데이터 소스로부터 데이터를 읽고 쓸 수 있도록 일반적인 함수를 제공
• 상위 수준 데이터 소스API는 DataFrameReader 및 DataFrameWriter는 서로 다른 데이터 소스간에 의사소통하는 방법을 제공

DataFrameReader

데이터 소스에서 데이터 프레임으로 데이터를 읽음

• 정의된 형식 및 권장되는 사용 패턴

DataFrameReader.format(args).option(“key”,”value”).schema(args).load()

오직 SparkSession 인스턴스를 통해서만 DataFrameReader에 액세스 할 수 있음

즉, DataFrame의 인스턴스를 개별적으로 만들 수는 없음

다음 코드를 이용해 인스턴스 핸들을 얻음

SparkSession.read

SparkSession.readStream

read는 정적 데이터 소스에서 DataFrame으로 읽기 위해 DataFrameReader에 대한 핸들을 반환

반면, readStream은 스트리밍 소스에서 읽을 인스턴스를 반환

 

---

 

DataFrameReader 함수, 인수 및 옵션

 

• 일반적으로 정적 파케이 데이터 소스에서 읽을때는 스키마가 필요하지 않음.
• 그러나 스트리밍 데이터 소스의 경우에는 스키마를 제공해야 함
• 파케이는 효율적이고 칼럼 기반 스토리지를 사용하며 빠른 압축 알고리즘을 사용하기 때문에 스파크의 기본이자 선호하는 데이터 소스. 이후 카탈리스트 옵티마이저를 다를 때 추가 이점이 있음

DataFrameWriter

• 지정된 내장 데이터 소스에 데이터를 저장하거나 쓰는 작업을 수행한다.
• DataReader와 달리 SparkSession이 아닌 저장하려는 데이터프레임에서 인스턴스에 엑세스 가능하다.
• 권장되는 사용 형식 (python)

DataFrameWriter.format(args)
	.option(args)
	.bucketBy(args)
	.partitionBy(args)
	.save(path)

DataFrameWriter.format(args).option(args).bucketBy(args).partitionBy(args).save(path)

instance handling 시에는 scala에서 진행

DataFrame.write 
// or 
DataFrame.writeStream

그 외 함수 (DataFrameWriter 함수, 인수 및 옵션)

함수	인수	설명
format()	“parquet”, “csv”, “txt”, “json”, “jdbc”, “orc” , “avro” 등	지정하지 않으면 default는 parquet나 spark.sql.sources.default값으로 설정된다.
option()	(”mode”, {append : overwrite : ignore : error or errorIfExists} ) (”mode”, {SaveMode.Overwrite : SaveMode.Append, Save Mode.Ignore, SaveMode.ErrorIfExists}) (”path”, “path_to_wirte_to”)	키/값 페어 및 옵션. 기본 모드 옵션은 error 또는 errorIfExists와 SaveModel이다. ErrorIfExists 는 데이터가 이미 있을 경우 런타임에서 예외를 발생시킨다.
bucketBy()	(numBuckets, col, col,…,coln)	버킷 개수 및 버킷 기준 칼럼 이름. 파일 시스템에서 하이브의 버킷팅 체계를 사용.
save()	“/path/to/data/source”	데이터 스스의 경로. option(”path”, “…”)에 지정된 경우 비어 있을 수 있다.
saveAsTable()	“table_name”	저장할 테이블

ex) json 사용하는 경우

// scala 예시
val location = ...
df.write.format("json").mode("overwrite").save(location)

파케이 파일 읽어오기 (데이터프레임, SQL테이블)

• 파케이 : 스파크의 기본 데이터 소스. 저장 공간 절약과 빠른 엑세스를 허용하는 압축을 제공하는 오픈소스 칼럼 기반 파일 형식
• 데이터 프레임으로 읽는 경우 code

file = """/data_path"""
df = spark.read_format("parquet").load(file) 
# 이때, 파케이가 메타데이터의 일부를 저장하기 때문에 스키마를 제공할 필요가 없다. (단, 스트리밍 데이터 소스에서 읽는 경우 제외)

• Spark SQL Table로 읽기

spark.sql("SELECT * FROM us_delay_flights_tbl").show()

데이터 프레임을 파케이 파일로 쓰기

DataFrameWriter에 대한 함수와 인수를 사용해서 파케이 파일을 저장할 위치를 제공하면 된다.

‘overwrite’는 데이터프레임을 저장할 떄 이미 해당 경로에 파일이 존재하면 덮어쓰기

‘compression’은 압축 방식을 지정하는 옵션, 압축 방식으로는 gzip, snappy, lz4, bzip2, zstd 등이 있다.

# In Python
(df.write.format("parquet")
    .mode("overwrite")
    .option("compression", "snappy")
    .save("/tmp/data/parquet/df_parquet"))

결과적으로 snappy라는 이름의 압축파일이 생성된다. 

스파크 SQL 테이블에 데이터 프레임 쓰기

saveAsTable()을 사용해 us_delay_flights_tbl 관리형 테이블이 생성된다.

# In Python
(df.write
	.mode("overwrite")
	.saveAsTable("us_delay_flights_tbl"))

파케이는 스파크에서 선호되는 기본 내장 데이터 소스 파일 형식이며 ETL 및 데이터 수집 프로세스에서 사용을 추천한다.

JSON(JavaScript Object Notation)

널리 사용되는 데이터 형식, XML에 비해 읽기 쉽고 분석하기 쉬운 형태이다.

단일 라인모드 - 각 라인이 단일 JSON 개체를 나타낸다.

다중 라인모드 - 전체 라인 객체는 단일 JSON 개체를 구성한다.

option()함수에서 multiLine을 true로 설정하면 읽을 수 있다.

JSON 파일을 데이터 프레임으로 읽기

피케이와 동일한 방식에 format() 함수만 더해 JSON 파일을 데이터 프레임으로 읽을 수 있다.

# In Python
file = "/Users/User/vscode/Spark/databricks-datasets/learning-spark-v2/flights/summary-data/json/*"
df = spark.read.format("json").load(file)

스파크 SQL 테이블로 JSON 파일 읽기

파케이에서 했던 것처럼 JSON 파일에서 SQL 테이블을 만들 수도 있다.

# -- In SQL
spark.sql("""
    CREATE OR REPLACE TEMPORARY VIEW us_delay_flights_tbl USING json
    OPTIONS (
    path "/Users/hyunjun/vscode/Spark/databricks-datasets/learning-spark-v2/flights/summary-data/json/*"
    )
""")

#// In Scala/Python
spark.sql("SELECT * FROM us_delay_flights_tbl").show()

 

데이터 프레임을 JSON 파일로 쓰기

DataFrameWriter 함수 및 인자를 지정하고 JSON 파일을 저장할 위치를 주면 JSON파일로 쓸 수 있다.

# In Python
(df.write.format("json")
    .mode("overwrite")
    .option("compression", "snappy")
    .save("/tmp/data/json/df_json"))

JSON 데이터 소스 옵션

• 다음 표 4-3는 DataFrameReader 및 DataFrameWriter에 대한 일반적인 JSON 옵션을 설명한다.

CSV

• csv 파일 형식은 쉼표로 각 데이터 또는 필더로 구분하며, 쉼표로 구분된 각 줄은 레코드를 나타낸다.
  • 쉼표 외 다른 구분 기호를 사용해 필드를 분리할 수 있다.

CSV 파일을 데이터 프레임으로 읽기

• DataFrameReader의 함수 및 인수를 사용해 CSV 파일을 데이터 프레임으로 읽을 수 있다.

csv_file = "../data/databricks-datasets/learning-spark-v2/flights/summary-data/csv/*"
schema = "DEST_COUNTRY_NAME STRING, ORIGIN_COUNTRY_NAME STRING, count INT"

df = (spark.read.format("csv")
      .option("header", "true")
      .schema(schema)
      .option("mode", "FAILFAST") 
      .option("nullValue", "")
      .load(csv_file))
      
spark.sql("SELECT * FROM us_delay_flights_tbl").show(10)

 

에이브로

• Avro(에이브로): 아파치 카프카에서 메시지를 직렬화 및 역직렬화할 때 사용
  • JSON에 대한 직접 매핑, 속도와 효율성, 바인딩을 포함한 여러 이점 제공

에이브로 파일을 데이터 프레임으로 읽기

• DataFrameReader를 사용하여 에이브로 파일을 데이터 프레임으로 읽는 것은 다른 데이터 소스에서 사용하는 방식과 다르지 않다

df = (spark.read.format('avro')
	.load('/databricks-datasets/learning-spark-v2/flights/summary-data/avro/*'))
df.show(truncate=False)

에이브로 파일을 스파크 SQL 테이블로 읽기

• 테이블 생성 (SQL로 진행)

CREATE OR REPLACE TEMPORARY VIEW episode_tbl
	USING avro
	OPTIONS (
	path "/databricks-datasets/learning-spark-v2/flights/summary-data/avro/*"
	)

• SQL을 사용하여 데이터 프레임으로 데이터 읽기

spark.sql("SELECT * FROM episode_tbl").show(truncate=False)

데이터 프레임을 에이브로 파일로 쓰기

(df.write
	.format("avro")
	.mode("overwrite")
	.save("/tmp/data/avro/df_avro"))
    
#지정된 위치에 압축 파일들로 채워진 폴더를 생성
rw-r--r-- 1 jules wheel 0 May 17 11:54 _SUCCESS
-rw-r--r-- 1 jules wheel 526 May 17 11:54 part-00000-ffdf70f4-<...>-c000.avro

에이브로 데이터 소스 옵션 (표)

ORC

• Spark의 2가지 설정으로 어떤 ORC 구현체를 사용할지 지정할 수 있음
  • spark.sql.orc.impl을 native로 설정
  • spark.sql.orc.enableVectorizedReader를 true로 설정
• 벡터화된 ORC 리더는 한번에 한 행을 읽는 것이 아닌 행 블록(블록 당 1024개)을 읽어 작업을 간소화하고 검색, 필터, 집계 및 조인과 같은 집중적인 작업에 대한 CPU 사용량을 줄임

ORC 파일을 데이터 프레임으로 읽기

file = ‘/databrics-datasets/learning-spark-v2/flights/summary-data/orc/*’
df = spark.read.format(’orc’).option(’path’, file).load()
df.show(10, False)

Reading an ORC file into a Spark SQL table

• ORC 데이터를 사용하여 SQL을 만들면 Parquet, JSON, CSV, Avro에서 차이가 없다
• ORC data 사용하여 Table 만들기

CREATE OR REPLACE TEMPORARY VIEW us_delay_flights_tbl
 USING orc
 OPTIONS (
 path "/databricks-datasets/learning-spark-v2/flights/summary-data/orc/*"
 )

• Scala/Python으로 data 조회하면

// In Scala/Python
spark.sql("SELECT * FROM us_delay_flights_tbl").show()

Writing DataFrames to ORC files

• DataFrame으로 변형된 것은 DataFrameWriter에서 사용하는 것과 같이 쓸 수 있다

// In Scala
df.write.format("orc")
 .mode("overwrite")
 .option("compression", "snappy")
 .save("/tmp/data/orc/df_orc")
# In Python
(df.write.format("orc")
 .mode("overwrite")
 .option("compression", "snappy")
 .save("/tmp/data/orc/flights_orc"))

⇒ 결과는 특정 위치에 ORC 파일로 압축되어 위치할 것이다.

Images

• Spark 2.4 커뮤니티에서 TensorFlow와 PyTorch와 같이 딥러닝과 머신러닝 프레임워크를 제공하는 새로운 데이터 파일 image files를 소개했다.

Reading an image file into a DataFrame

• 이전 file format과 같이 DataFrameReader 메소드와 옵션들로 image file를 읽을 수 있다.
•  

// In Scala
import org.apache.spark.ml.source.image
val imageDir = "/databricks-datasets/learning-spark-v2/cctvVideos/train_images/"
val imagesDF = spark.read.format("image").load(imageDir)
imagesDF.printSchema
imagesDF.select("image.height", "image.width", "image.nChannels", "image.mode",
 "label").show(5, false)

# In Python
from pyspark.ml import image
image_dir = "/databricks-datasets/learning-spark-v2/cctvVideos/train_images/"
images_df = spark.read.format("image").load(image_dir)
images_df.printSchema()
root
|-- image: struct (nullable = true)
| |-- origin: string (nullable = true)
| |-- height: integer (nullable = true)
| |-- width: integer (nullable = true)
| |-- nChannels: integer (nullable = true)
| |-- mode: integer (nullable = true)
| |-- data: binary (nullable = true)
|-- label: integer (nullable = true)
images_df.select("image.height", "image.width", "image.nChannels", "image.mode",
 "label").show(5, truncate=False)

Binary Files

• Spark 3.0은 binary file을 데이터 소스처럼 제공하도록 기능을 추가했다. (데이터 변환기는 각 binary file를 하나의 DataFrame 열로 포함한다)
• binary file data는 다음과 같은 기능을 제공한다

1. path : StringType
2. modificationTime: TimestampType
3. length: LongType
4. content: BinaryType

Reading a binary file into a DataFrame

• binary file을 읽기 위해서는 Data를 binaryFile 형식으로 지정해야한다.
• 데이터 소스 옵션인 pathGlobFilter를 사용하여 파티션 검색 동작을 유지하면서 지정된 전역 패턴과 일치하는 경로로 파일을 Load 할 수 있다.
• 파티션된 경로들에서 입력된 경로를 통해 JPG file를 읽는 코드

// In Scala
val path = "/databricks-datasets/learning-spark-v2/cctvVideos/train_images/"
val binaryFilesDF = spark.read.format("binaryFile")
 .option("pathGlobFilter", "*.jpg")
 .load(path)
binaryFilesDF.show(5)

# In Python
path = "/databricks-datasets/learning-spark-v2/cctvVideos/train_images/"
binary_files_df = (spark.read.format("binaryFile")
 .option("pathGlobFilter", "*.jpg")
 .load(path))
binary_files_df.show(5)

⇒ 결과 데이터 : recursiveFileLookup 옵션이 true 되어있는 경우 label column이 없다

이번 포스팅에서, 데이터를 DataFrame으로 어떻게 읽고 어떤 file format을 제공하는지 보고, 내장된 데이터 소스에서 어떻게 임시 view를 만들수 있는지 알아 보았다. DataFrame API 혹은 SQL을 사용하던지 결과는 같다.

Summary

요약하자면 이 장에서는 DataFrame API와 Spark SQL간의 상호 운영성을 보았다

• Spark SQL과 DataFrame API를 사용하여 관리와 비관리 table들 만들기
• 다양한 내장 데이터 소스 및 파일 읽고 쓰기
• spark.sql 를 사용하여 Spark SQL 테이블 및 뷰로 저장된 구조화된 SQL 쿼리문 실행
• Spark Catalog를 통해 테이블 및 뷰 (메타데이터와 관련된) 검사
• DataFrameWriter 및 DataFrameReader API 사용