PySpark資料處理與視覺化技術

# 使用withColumn函式新增ratings欄位
df_with_newcols = df_with_newcols.withColumn('ratings', 
                                            when(df['popularity'] < 3, 'Low')
                                            .when(df['popularity'] < 5, 'Mid')
                                            .otherwise('High'))

# 將budget_cat和ratings欄位合併為新的欄位BudgetRating_Category
df_with_newcols = df_with_newcols.withColumn('BudgetRating_Category', 
                                            concat(df_with_newcols.budget_cat, df_with_newcols.ratings))

# 將新的欄位轉換為小寫並去除空白
df_with_newcols = df_with_newcols.withColumn('BudgetRating_Category', 
                                            trim(lower(df_with_newcols.BudgetRating_Category)))

# 將DataFrame註冊為臨時表
df_with_newcols.createOrReplaceTempView('temp_data')

# 使用SQL查詢來統計ratings的分佈
spark.sql('SELECT ratings, COUNT(ratings) FROM temp_data GROUP BY ratings').show()

# 匯入視窗函式相關的模組
from pyspark.sql.window import Window
from pyspark.sql.functions import ntile

# 過濾掉popularity為空或NaN的記錄
df_with_newcols = df_with_newcols.filter((df_with_newcols['popularity'].isNotNull()) & (~isnan(df_with_newcols['popularity'])))

# 使用ntile函式計算popularity的十分位數
df_with_newcols = df_with_newcols.select("id", "budget", "popularity",
                                        ntile(10).over(Window.orderBy(df_with_newcols['popularity'].desc())).alias("decile_rank"))

# 對decile_rank進行分組統計
df_with_newcols.groupby("decile_rank").agg(min('popularity').alias('min_popularity'),
                                          max('popularity').alias('max_popularity'),
                                          count('popularity')).show()

# 步驟1：匯入視窗函式相關模組
from pyspark.sql.window import Window
from pyspark.sql.functions import year, rank

# 步驟2：選擇需要的欄位子集
df_second_best = df.select('id', 'popularity', 'release_date')

# 步驟3：從發布日期建立年份欄位
df_second_best = df_second_best.withColumn('release_year', year('release_date')).drop('release_date')

# 步驟4：定義視窗函式
year_window = Window.partitionBy(df_second_best['release_year']).orderBy(df_second_best['popularity'].desc())

# 步驟5：應用視窗函式進行排名
df_second_best = df_second_best.select('id', 'popularity', 'release_year', rank().over(year_window).alias("rank"))

# 步驟6：找出1970年排名第二的電影
df_second_best.filter((df_second_best['release_year'] == 1970) & (df_second_best['rank'] == 2)).show()

# 步驟1：匯入必要的模組
from pyspark.sql.window import Window
from pyspark.sql.functions import year, max

# 步驟2：選擇需要的欄位
df_revenue = df.select('id', 'revenue', 'release_date')

# 步驟3：建立年份欄位
df_revenue = df_revenue.withColumn('release_year', year('release_date')).drop('release_date')

# 步驟4：定義包含範圍的視窗函式
windowRev = Window.partitionBy(df_revenue['release_year']).orderBy(df_revenue['revenue'].desc()).rangeBetween(-sys.maxsize, sys.maxsize)

# 步驟5：計算收入差異
revenue_difference = max(df_revenue['revenue']).over(windowRev) - df_revenue['revenue']

# 步驟6：顯示最終結果
df_revenue.select('id', 'revenue', 'release_year', revenue_difference.alias("revenue_difference")).show(10, False)

# 步驟1：建立年份欄位
df = df.withColumn('release_year', year('release_date'))

# 步驟2：應用collect_list函式收集所有出現的年份
df.filter("title=='The Lost World'").groupby('title').agg(collect_list("release_year")).show(1, False)

df_sample = df.sample(False, 0.4, 11)
df_sample.count()

df_sample = df.sample(True, 0.4, 11)
df_sample.count()

df_strat = df.sampleBy("release_year", fractions={1959: 0.2, 1960: 0.4, 1961: 0.4}, seed=11)
df_strat.count()

df.cache()

df.write.format('csv').option('delimiter', '|').save('output_df')

df.coalesce(1).write.format('csv').option('delimiter', '|').save('output_coalesced_df')

df.write.partitionBy('release_year').format('csv').option('delimiter', '|').save('output_df_partitioned')

df.write.saveAsTable('film_ratings')

df_pandas = df.toPandas()

df_py = spark.createDataFrame(df_pandas)

df.join(df_p1, df['id'] == df_p1['id'], 'inner').printSchema()

df.join(df_p1, df['id'] == df_p1['id'], 'left').count()

df.join(df_p1, df['id'] == df_p1['id'], 'right').count()

df.join(df_p1, df['id'] == df_p1['id'], 'outer').count()