Spark 데이터처리 실습 2 (컬럼명과 타입 추가하기 + 정규표현식 + Pandas와 비교)- TIL230704

📚 KDT WEEK 14 DAY 2 TIL

• Pandas로 처리하기
• Spark로 처리하기
  • Spark 환경 설정 및 csv 파일 처리하기
  • 데이터 프레임에 함수 사용해보기
  • Spark SQL로 처리해보기
  • 알아두면 좋은 기능

 

 

---

 

이전 글에서 Spark 데이터처리 실습을 해보았다.

https://sunhokimdev.tistory.com/60

Spark 데이터처리 실습 - TIL230704

 

저번과 같은 환경을 사용한다.

Colab에서 입력할 때는 제일 앞에 ! 을 붙이면 된다.

pip install pyspark==3.3.1 py4j==0.10.9.5

 

사용한 데이터(1800.csv)

 

간단한 csv 파일로, 헤더가 없는 형태이다. 코드에서 컬럼명을 따로 추가할 것이다.

 

 

---

 

Pandas 처리 vs Spark 처리

 

🟥 Pandas로 처리하는 경우

 

1800.csv 파일을 불러와서 몇 번의 과정을 통해 필요한 데이터를 얻어낸다.

 

import pandas as pd 

pd_df = pd.read_csv(
    "1800.csv",
    names=["stationID", "date", "measure_type", "temperature"], # 헤더가 없으므로 이름 지정
    usecols=[0, 1, 2, 3] # 처음 네개의 컬럼만 읽는다
)

pd_df.head() # 데이터 확인

# measure_type 컬럼이 TMIN인 데이터만 필터링
pd_minTemps = pd_df[pd_df['measure_type'] == "TMIN"] 

# 두 개의 컬럼만 선택
pd_stationTemps = pd_minTemps[["stationID", "temperature"]] 

# stationID로 Groupby + temperature가 가장 작은 것
pd_minTempsByStation = pd_stationTemps.groupby(["stationID"]).min("temperature")

 

수행 결과

 

---

 

🟦 Spark로 처리하기

 

1) spark 환경 설정 및 csv 파일 로드하기

 

csv 파일을 로드한 뒤, 컬럼을 지정해주었다.

from pyspark.sql import SparkSession
from pyspark import SparkConf

# Spark 환경 설정
conf = SparkConf()
conf.set("spark.app.name", "PySpark DataFrame #1")
conf.set("spark.master", "local[*]") # Google Colab의 CPU 수만큼 할당

spark = SparkSession.builder\
        .config(conf=conf)\
        .getOrCreate()
        
# csv 파일로드하기 (컬럼, 타입지정 없이) > 타입이 모두 string이됨
df = spark.read.format("csv").load("1800.csv") # spark.read.csv("1800.csv")

# csv 파일로드하기 (하나씩 컬럼지정하기)
df = spark.read.format("csv")\
    .load("1800.csv")\
    .toDF("stationID", "date", "measure_type", "temperature", "_c4", "_c5", "_c6", "_c7")
    
# .option으로 스키마를 추측하라고 호출하면, spark이 레코드 몇 개를 보고 추측해서 타입 넣어줌
df = spark.read.format("csv")\
    .option("inferSchema", "true")\
    .load("1800.csv")\
    .toDF("stationID", "date", "measure_type", "temperature", "_c4", "_c5", "_c6", "_c7")

 

각 컬럼별 타입에 집중해보자.

 

각각 스키마를 출력해봤을 때, 컬럼명과 타입이 다름을 비교해보자. 

타입을 명시해주지 않으면 그냥 String 타입으로 붙여버린다.

하지만 csv 파일을 로드하는 과정에서 option으로 inferSchema를 주면, 알아서 추측해준다.

 

추가적인 방법으로, 명시적으로 타입을 알려주는 방법도 있다.

 

 

이렇게 명시해주는 방법이 확실한 것 같다.

 

명시해줄 때 사용하는 타입으로는 다음의 링크를 참고하자.

pyspark.sql.types의 타입 리스트 : https://spark.apache.org/docs/latest/sql-ref-datatypes.html

Data Types - Spark 3.4.1 Documentation

 

 

2) 데이터 프레임에 함수 사용해보기

 

Pandas에서 데이터를 처리할 때처럼 부분적으로 처리한다.

# measure_type = "TMIN" 값을 찾는 세 가지 방법(결과 동일)
minTemps = df.filter(df.measure_type == "TMIN")
minTemps = df.where(df.measure_type == "TMIN")
minTemps = df.where("measure_type = 'TMIN'") # SQL where 조건절 명시하듯이

# stationID 컬럼으로 GroupBy + temperature의 최솟값에 해당하는 데이터 찾기
minTempsByStation = minTemps.groupBy("stationID").min("temperature")

# 두 가지 컬럼을 지정해서 SELECT하는 두 가지 방법(결과 동일)
stationTemps = minTemps[["stationID", "temperature"]]
stationTemps = minTemps.select("stationID", "temperature")

# 결과 출력해보기
results = minTempsByStation.collect()
for result in results:
    print(result[0] + "\t{:.2f}F".format(result[1]))

 

 

결과

 

3) Spark SQL로 처리하기

 

간단하게 SQL 쿼리문을 사용해서, 어렵지 않게 처리할 수 있다.

 

4) 알아두면 좋은 기능!

 

a) agg 함수 컬럼에 이름 지정하기(withColumnRenamed, sql.functions)

 

# 뒤에 withColumnRenamed 사용
df_ca = df.groupBy("cust_id").sum("amount_spent").withColumnRenamed("sum(amount_spent)", "sum")

# sql.functions 사용해서 간단하게 여러 개 명시하기
import pyspark.sql.functions as f
df.groupBy("cust_id") \
   .agg(
       f.sum('amount_spent').alias('sum'),
       f.max('amount_spent').alias('max'),
       f.avg('amount_spent').alias('avg')).collect()

 

b) 텍스트 파일 불러와서 Spark로 regex(정규표현식) 적용해서 추출해보기

 

transfer_cost.txt(일부)

On 2021-01-04 the cost per ton from 85001 to 85002 is $28.32 at ABC Hauling
On 2021-01-04 the cost per ton from 85001 to 85004 is $25.68 at ABC Hauling
On 2021-01-04 the cost per ton from 85001 to 85007 is 19.86 at ABC Hauling
On 2021-01-04 the cost per ton from 85001 to 85007 is 20.52 at Haul Today
On 2021-01-04 the cost per ton from 85001 to 85010 is 20.72 at Haul Today

 

# Spark 설정은 위 실습과 동일

import pyspark.sql.functions as F
from pyspark.sql.types import *

schema = StructType([ StructField("text", StringType(), True)])
transfer_cost_df = spark.read.schema(schema).text("transfer_cost.txt") # txt 파일이므로 ,text 사용

transfer_cost_df.show(truncate=False) # truncate=False 사용해서 잘리는 거 없이 일부(20개)를 출력

# regex 적용해서 text 내용을 각 컬럼으로 넣기
from pyspark.sql.functions import *
regex_str = r'On (\S+) the cost per ton from (\d+) to (\d+) is (\S+) at (.*)'

# withColumn("추가하거나 존재하는 컬럼이름", "채워질 값")
# regexp_extract("추출할 컬럼", "타입", 매칭되는 것중 몇 번째인지(1부터시작))
df_with_new_columns = transfer_cost_df\
    .withColumn('week', regexp_extract('text', regex_str, 1))\
    .withColumn('departure_zipcode', regexp_extract(column('text'), regex_str, 2))\
    .withColumn('arrival_zipcode', regexp_extract(transfer_cost_df.text, regex_str, 3))\
    .withColumn('cost', regexp_extract(col('text'), regex_str, 4))\
    .withColumn('vendor', regexp_extract(col('text'), regex_str, 5))
    
final_df = df_with_new_columns.drop("text") # 기존의 text 컬럼은 drop

 

 

c) 내용을 csv과 json으로 저장해보기

 

# 로컬에 csv 파일로 저장하지만, 폴더로 저장됨
final_df.write.csv("extracted.csv")

# json 파일로 저장하지만, 폴더로 저장됨
final_df.write.format("json").save("extracted.json")

 

 

원래 빅데이터 처리용이므로, 폴더로 저장되어 데이터 블록단위로 폴더 안에 나누어 저장된다.

실습 데이터는 그렇게 크지 않아서 하나로만 저장되었고, 그 안을 확인해보면 잘 저장된 것을 확인할 수 있다.