Spark Tuning examples

1. 캐싱

• 캐싱은 클러스터의 익스큐터 전반에 걸쳐 만들어진 임시저장소(메모리나 디스크)에 DataFrame, 테이블 또는 RDD를 보관해 빠르게 접근할 수 있도록 합니다.
• 직렬화, 역직렬화, 그리고 저장소 자원을 소모하기 때문에 항상 유용하지는 않습니다.
• 캐시할 데이터 저장소 레벨을 지정할 수 있습니다. (ex. MEMORY_AND_DISK)

df1 = spark.read.format("csv")
                .option("inferSchema", "true")
                .option("header", "true")
                .load("/data/sample.csv")
# df1.cache()
df1.persist(StorageLevel.MEMORY_AND_DISK)
df1.count() # ACTION

df2 = df1.groupBy("A").count().collect()
df3 = df1.groupBy("B").count().collect()
df4 = df1.groupBy("C").count().collect()

# df1.unpersist()
df1.()
# spark.catalog.clearCache() # 모든 캐시 제거

2. 브로드캐스트 조인

SELECT /*+ BROADCAST(Table 2) */ COLUMN
FROM Table 1 join Table 2
on Table1.key= Table2.key

3. 동일한 테이블은 최소한으로 읽어오는 것이 좋습니다.

• spark에서 어떤 칼럼을 기준으로 partitioning, bucketing 어있는지에 따라 동일한 테이블을 여러번 스캔하는 것이 더 좋을 수 있습니다.

[table 두번 스캔]
SELECT t.ym, t.cust_no, a.purchase_amt
FROM table t
where 1=1
and t.ym = (select max(ym) from table)

[table 한번 스캔]
select t.ym, t.cust_no, a.purchase_amt
from (select t.ym, t.cust_no, t.purchase_amt
            ,row_number() over (order by t.ym desc) as idx  
      from table t
where 1=1 
and t.idx = 1

4. Spark에서 select case when 안에 기간을 쓰는 경우

• 일반 DB(mysql 등)에서는 모든 레코드에 대해 case when 조건을 체크하지만,
• 동일한 데이터를 가진 테이블을 두번 스캔하는 것이 훨씬 느립니다.
• 하지만 yearmonth를 기준으로 partitioning 되어있는 파일시스템에서 spark로 데이터를 읽어온다면, 두 개의 쿼리로 나누더라도 동일한 데이터를 읽어오지 않습니다.
• 따라서 UNION ALL 하는 것이 훨씬 빠릅니다.

-- MYSQL 기준
-- CASE WHEN
select rental_id
	  ,case when return_date between str_to_date('20050501','%Y%m%d%H%i%s') and str_to_date('20050531','%Y%m%d%H%i%s') then 'Y' else 'N' end as 'yn5'
      ,case when return_date between str_to_date('20050601','%Y%m%d%H%i%s') and str_to_date('20050630','%Y%m%d%H%i%s') then 'Y' else 'N' end as 'yn6'
from sakila.rental

-- UNION ALL
(select rental_id
	   ,'Y' as 'yn5'
       ,'N' as 'yn6'
from sakila.rental
where 1=1
and extract(year_month from return_date) = '200505') 
union all
(select rental_id
       ,'N' as 'yn5'
       ,'Y' as 'yn6'
from sakila.rental
where 1=1
and extract(year_month from return_date) = '200506')

5. Semi Join

• 두 테이블을 조인할 때 한쪽 테이블이 작은 경우에 활용할 수 있습니다.
• 예를 들어 특정 조건에서 구매 실적이 있는 고객을 추출할 때 유용합니다. (특정 조건에 의해 테이블의 크기가 줄어들기 때문입니다.)

select count(*)
from (
     select t1.cust_no, t1.birthyear, t1.join_dt
     from cust_tb t1
     inner join (select distinct cust_no
                 from purchase_tb
                 where 1=1
                 and yyyymm >= '201901'
                 and store_cd = '001') t2
     on t1.cust_no = t2.cust_no
) t

select count(*)
from(
    select t1.cust_no, t1.birthyear, t1.memb_join_dt
    from cust_tb t1
    where exists (select '1'
                  from purchase_tb t2
                  where 1=1
                  and t1.cust_no = t2.dpstr_cust_no
                  and t2.yyyymm >= '201901'
                  and store_cd = '0001')
)t