17. SparkSQL(JOIN - DataFrame, SQL)

실습 문제들

---

1. JOIN 실습

2. 매출 사용자 10명 알아내기(RANKING)

3. 월별 채널별 매출과 방문자 정보 계산하기(Grouping)

4. 사용자별로 처음 채널과 마지막 채널 알아내기(Windowing)

 

실습 사용 테이블 3개 설명 - 사용자 ID, 세션 ID

• 사용자 ID :
  • 보통 웹서비스에서는 등록된 사용자마다 유일한 ID를 부여 -> 사용자 ID
• 세션 ID :
  • 사용자가 외부 링크(보통 광고)를 타고 오거나 직접 방문해서 올 경우 세션을 생성
  • 즉 하나의 사용자 ID는 여러 개의 세션 ID를 가질 수 있음
  • 보통 세션의 경우 세션을 만들어낸 소스를 채널이란 이름으로 기록해둠
    • 마케팅 관련 기여도 분석을 위함
  • 또한 세션이 생긴 시간도 기록
• 이 정보를 기반으로 다양한 데이터 분석과 지표 설정이 가능
  • 마케팅 관련
  • 사용자 트래픽 관련
• 채널과 채널 기여도
  • 사용자 ID 100번 : 총 3개의 세션(파란 배경)을 갖는 예제
    • 세션 1: 구글 키워드 광고로 시작한 세션
    • 세션 2: 페이스북 광고를 통해 생긴 세션
    • 세션 3: 네이버 광고를 통해 생긴 세션
  • 이를 통해 매출 기여도 분석을 할 수 있다
    • FIRST TOUCH : 상품 구매를 첫번째 구글 광고를 통해 한 것(CREDIT)
    • LAST TOUCH : 직전 채널에(네이버 광고에 크레딧)
    • MULTI TOUCH : 상품 구매에 관여한 모든 광고에 크레딧(가장 최근에 준 것에 크레딧을 주기도함)

---

1. JOIN 실습(DataFrame)

• 앞서 예제로 배운 두 개의 테이블 대상
  • Vital
  • Alert
• 6개의 조인 수행
  • INNER
  • LEFT
  • RIGHT
  • FULL
  • CROSS
  • SELF

(1) 실습 테이블 2개 로딩

vital = [
     { 'UserID': 100, 'VitalID': 1, 'Date': '2020-01-01', 'Weight': 75 },
     { 'UserID': 100, 'VitalID': 2, 'Date': '2020-01-02', 'Weight': 78 },
     { 'UserID': 101, 'VitalID': 3, 'Date': '2020-01-01', 'Weight': 90 },
     { 'UserID': 101, 'VitalID': 4, 'Date': '2020-01-02', 'Weight': 95 },
]

alert = [
    { 'AlertID': 1, 'VitalID': 4, 'AlertType': 'WeightIncrease', 'Date': '2020-01-01', 'UserID': 101},
    { 'AlertID': 2, 'VitalID': None, 'AlertType': 'MissingVital', 'Date': '2020-01-04', 'UserID': 100},
    { 'AlertID': 3, 'VitalID': None, 'AlertType': 'MissingVital', 'Date': '2020-01-05', 'UserID': 101}
]

rdd_vital = spark.sparkContext.parallelize(vital)
rdd_alert = spark.sparkContext.parallelize(alert)

df_vital = rdd_vital.toDF()
df_alert = rdd_alert.toDF()

df_vital.printSchema()
df_alert.printSchema()

 

(2) Join by DataFrame

# INNER JOIN
join_expr = df_vital.VitalID == df_alert.VitalID
df_vital.join(df_alert, join_expr, "inner").show()

# LEFT JOIN
join_expr = df_vital.VitalID == df_alert.VitalID
df_vital.join(df_alert, join_expr, "left").show()

# RIGHT JOIN
join_expr = df_vital.VitalID == df_alert.VitalID
df_vital.join(df_alert, join_expr, "right").show()

# FULL OUTER JOIN
join_expr = df_vital.VitalID == df_alert.VitalID
df_vital.join(df_alert, join_expr, "full").show()

# CROSS JOIN
df_vital.join(df_alert, None, "cross").show()

# SELF JOIN
join_expr = df_vital.VitalID == df_vital.VitalID
df_vital.join(df_vital, join_expr, "left").show()

---

2. JOIN 실습(SQL)

(1) 실습 테이블 2개 임시 테이블 생성

df_vital.createOrReplaceTempView("Vital")
df_alert.createOrReplaceTempView("Alert")

(2) Join by SQL

# INNER JOIN
df_inner_join = spark.sql("""SELECT * FROM Vital v
JOIN Alert a ON v.vitalID = a.vitalID;""")
df_inner_join.show()

# LEFT JOIN
df_left_join = spark.sql("""SELECT * FROM Vital v
LEFT JOIN Alert a ON v.vitalID = a.vitalID;""")
df_left_join.show()

# RIGHT JOIN
df_right_join = spark.sql("""SELECT * FROM Vital v
RIGHT JOIN Alert a ON v.vitalID = a.vitalID;""")
df_right_join.show()

# OUTER JOIN
df_outer_join = spark.sql("""SELECT * FROM Vital v
FULL JOIN Alert a ON v.vitalID = a.vitalID;""")
df_outer_join.show()

# CROSS JOIN
df_cross_join = spark.sql("""SELECT * FROM Vital v
CROSS JOIN Alert a""")
df_cross_join.show()

# self JOIN
df_self_join = spark.sql("""SELECT * FROM Vital v1
JOIN Vital v2""")
df_self_join.show()