1ambda

# dfConverted.select("education").limit(5).explain("extended")

== Analyzed Logical Plan ==
education: string
GlobalLimit 5
+- LocalLimit 5
   +- Project [education#238]
      +- Project [id#236, year_birth#237, education#238, count_kid#239, count_teen#240, date_customer#241, days_last_login#242, add_months(to_date('date_customer, Some(d-M-yyyy)), 72) AS date_joined#257]
         +- Project [ID#16 AS id#236, Year_Birth#17 AS year_birth#237, Education#18 AS education#238, Kidhome#21 AS count_kid#239, Teenhome#22 AS count_teen#240, Dt_Customer#23 AS date_customer#241, Recency#24 AS days_last_login#242]
            +- Relation[ID#16,Year_Birth#17,Education#18,Marital_Status#19,Income#20,Kidhome#21,Teenhome#22,Dt_Customer#23,Recency#24,MntWines#25,MntFruits#26,MntMeatProducts#27,MntFishProducts#28,MntSweetProducts#29,MntGoldProds#30,NumDealsPurchases#31,NumWebPurchases#32,NumCatalogPurchases#33,NumStorePurchases#34,NumWebVisitsMonth#35,AcceptedCmp3#36,AcceptedCmp4#37,AcceptedCmp5#38,AcceptedCmp1#39,... 5 more fie

1ambda

df.explain("cost")
df.explain("codegen")

1ambda

spark.sparkContext._conf.get('spark.default.parallelism')

200 # 출력 결과, Spark 설정에 따라 200 이 아닌 값일 수 있습니다.

1ambda

print(f"Partition Count of Dataframe df:\t\t{df.rdd.getNumPartitions()}")
print(f"Partition Count of Dataframe dfSelected:\t{dfSelected.rdd.getNumPartitions()}")
print(f"Partition Count of Dataframe dfConverted:\t{dfConverted.rdd.getNumPartitions()}")

# 출력 결과
Partition Count of Dataframe df:		1
Partition Count of Dataframe dfSelected:	1
Partition Count of Dataframe dfConverted:	1

1ambda

# repartition 함수를 통해 파티션 숫자를 1 -> 5 로 늘립니다.
dfPartitioned = dfConverted.repartition(5)

print(f"Partition Count of Dataframe dfPartitioned:\t{dfPartitioned.rdd.getNumPartitions()}")

Partition Count of Dataframe dfPartitioned:	5

1ambda

dfConverted.repartition(col("id"))

1ambda

from pyspark.sql.functions import *
from pyspark.sql.types import *
from pyspark.sql import Row

# DataBricks 로 실습한다면 경로를 "/FileStore/tables/marketing_campaign.csv" 로 변경합니다
df = spark.read.load("./marketing_campaign.csv",
                     format="csv", 
                     sep="\t", 
                     inferSchema="true", 
                     header="true")

1ambda

# 'collect()' 는 Executor 에서 파일 내의 데이터를 읽어 Driver 로 전송하는 Action 입니다.
# 만약 cache() 등을 통해 캐싱되어 있다면 메모리에서 데이터를 찾아 보낼 수 있습니다.
collected = dfPartitioned.collect()

# type(collected) 의 실행 결과
list

# collected[0] 의 실행 결과
Row(id=7196, year_birth=1950, education='PhD', count_kid=1, count_teen=1, date_customer='08-02-2014', days_last_login=20, date_joined=datetime.date(2020, 2, 8))

1ambda

from pyspark.sql import Row

missing_days = 10

# Spark 의 Row 는 read-only 입니다. 따라서 Python 에서 변경하기 위해 Dict 로 변경 후 다시 Row 로 되돌립니다.    
# 효율적인 방법이 아니며, 내부 동작의 이해를 돕기 위해 만든 코드입니다.    
def updateDaysLastLogin(row):    
    parsed = row.asDict()  
    parsed['days_last_login'] = parsed['days_last_login'] +  missing_days    

1ambda

spark.driver.cores # Driver 에서 사용할 CPU Core 숫자
spark.driver.memory # Driver 에서 사용할 메모리 GiB