Spark SQL 실습 1

실습

 

실습 1. 총 매출이 가장 많은 사용자 10명 찾기

• 테이블을 데이터프레임으로 로딩
• 데이터프레임별 테이블 이름 지정

 

!cd /usr/local/lib/python3.8/dist-packages/pyspark/jars && wget https://s3.amazonaws.com/redshift-downloads/drivers/jdbc/1.2.20.1043/RedshiftJDBC42-no-awssdk-1.2.20.1043.jar

from pyspark.sql import SparkSession

spark = SparkSession \
    .builder \
    .appName("Python Spark SQL #1") \
    .getOrCreate()

# Redshift와 연결 및 DataFrame으로 로딩하기
url = "jdbc:redshift://learnde.cduaw970ssvt.ap-northeast-2.redshift.amazonaws.com:5439/dev?user=guest&password=Guest1234"

df_user_session_channel = spark.read \
    .format("jdbc") \
    .option("driver", "com.amazon.redshift.jdbc42.Driver") \
    .option("url", url) \
    .option("dbtable", "raw_data.user_session_channel") \
    .load()

df_session_timestamp = spark.read \
    .format("jdbc") \
    .option("driver", "com.amazon.redshift.jdbc42.Driver") \
    .option("url", url) \
    .option("dbtable", "raw_data.session_timestamp") \
    .load()

df_session_transaction = spark.read \
    .format("jdbc") \
    .option("driver", "com.amazon.redshift.jdbc42.Driver") \
    .option("url", url) \
    .option("dbtable", "raw_data.session_transaction") \
    .load()
    
    
df_user_session_channel.createOrReplaceTempView("user_session_channel")
df_session_timestamp.createOrReplaceTempView("session_timestamp")
df_session_transaction.createOrReplaceTempView("session_transaction")

top_rev_user_df = spark.sql("""
    SELECT userid,
        SUM(str.amount) revenue,
        SUM(CASE WHEN str.refunded = False THEN str.amount END) net_revenue
    FROM user_session_channel usc
    JOIN session_transaction str ON usc.sessionid = str.sessionid
    GROUP BY 1
    ORDER BY 2 DESC
    LIMIT 10
    """)
    
top_rev_user_df2 = spark.sql("""
    SELECT
    	userid,
    	SUM(amount) total_amount, 
        	RANK() OVER (ORDER BY SUM(amount) DESC) rank
    FROM session_transaction st
    JOIN user_session_channel usc ON st.sessionid = usc.sessionid
    GROUP BY userid
    ORDER BY rank
    LIMIT 10
""")