pyspark 및 spark 꿀팁 정리(feat.Databricks)

컬럼에 한국어 사용하기

---

select 
    sda.sell_daddr_id `배송ID`
from dw.sell_deliy_addr sda

• `컬럼명`을 이용해서 한국어로 된 컬럼을 사용할 수 있음

 

데이터프레임 요약해서 보기

---

dbutils.data.summarize(데이터프레임)

• Dataframe 의 정보를 요약해서 보여줌
• missing (결측치), mean, zeros, min, median, max 등등 여러 값도 한눈에 볼 수 있음

 

dbfs(DataBricks File System)

---

dbutils.fs.ls("/")

• dbfs 에 위치하는 파일을 보여줌
• 이는 현재 S3 의 저 위치에 있는 파일의 정보를 보여줌
  hive metastore 의 스키마는 어디에 위치하는가?
• 참고 url

Databricks documentation

 

결측치 제거하기

---

values_to_fill = {
    "배송방법": "Unknown",
    "운송사": "Unknown",
    "송장번호": "Unknown",
    "지역구분값": "Unknown",
    "도착국가": "Unknown",
    "추천박스": "Unknown",
    "실제포장박스": "Unknown",
    "추천박스부피": -1.0,
    "배송총무게": -1.0,
    "배송총실제무게": -1.0,
    "추천박스부피": -1.0,
    "실제박스부피": -1.0,
    "환율": -1.0,
    "총배송비부가세": 0,
    "총원화배송비": 0.0,
    "총원화배송비부가세": 0.0,
    "총배송비수익": 0.0,
}

df_filled = df_transformed.fillna(values_to_fill)

• values_to_fill 은 결측치를 어떻게 표시할 지에 대해 key: value 로 표기함
• 각 value 에 해당하는 것은 데이터프레임.printSchema() 으로 컬럼이 무슨 Type 인지 확인하고 해당 컬럼이 절대 가지지 않을 값 등으로 대체함

 

string to json

---

import pyspark.sql.functions as F
from pyspark.sql.types import MapType, StringType, ArrayType

convert_explode_df = df.withColumn(
    "base_parms_json",
    F.from_json(F.col("base_parms"), MapType(StringType(), StringType())), 
    # base_parms 에서 {"key":value} 형태의 String 을 String:String 형태의 데이터타입으로 변환
)

• pyspark.sql.types 에 있는 MapType 과 StringType 을 이용해 범용적으로 바꿀 수 있음
• 어떤 값이 나올지 모를 땐 위와 같은 방식으로 하는 게 더 좋다고 생각함

 

from pyspark.sql.functions import from_json 
from pyspark.sql.types import StructType, IntegerType, StringType

# 이미 정의된 스키마 
json_schema = StructType() \
                .add("age", IntegerType()) \
                .add("city", StringType())

df_with_parsed_column = df.withColumn("parsed_json", from_json(df["jsonData"], json_schema))

• 혹은 StructType() 을 이용해서 변환할 수도 있음

 

from_json(컬럼명, schema_of_json('json 예시')

• 이렇게 사용할 경우, 스키마를 유추하여 사용하기도 함

 

 

데이터프레임을 컬럼명으로 정렬

---

org_df_sorted = org_df.select(sorted(org_df.columns))
df_transformed_sorted = df_transformed.select(sorted(df_transformed.columns))

• 컬럼명을 기준으로 데이터프레임의 데이터를 재배치하고 싶을 경우에 사용함

 

스키마 및 테이블 상세 정보보기

describe detail 스키마명.테이블명
describe history 스키마명.테이블명
describe extended 스키마명.테이블명
describe database extended 스키마명

 

Delta Log 보기

---

f = dbutils.fs.head("경로")
print(f)

 

 

Spark SQL 로 JSON 분해해서 보기

---

select json인 컬럼명:key
from 스키마명.테이블명

# 예시
select *, base_parms:SHOP_NO
from dw.user_action_history

 

 

SQL UDFs

---

CREATE FUNCTION 함수명(파라미터명 타입)
RETURNS 타입
RETURN CASE ~~

# ex)
CREATE FUNCTION site_type(email STRING)
RETURNS STRING
RETURN CASE
            WHEN email like "%.com" THEN "Commercial business"
            WHEN email like "%.org" THEN "Non-profits organization"
            WHEN email like "%.edu" THEN "Educational institution"
            ELSE concat("Unknow extensino for domain: ", split(email, "@")[1])
        END;