Spark学习 day8 - 呓语

对各省销售指标，每个省的销售额进行统计
TOP3销售省份中，有多少家店铺日均销售额 > 1000
TOP3省份中各个省份的平均单单价
TOP3省份中，各个省份的支付类型比例

#coding:utf8
from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.storagelevel import StorageLevel
from  pyspark.sql.types import StringTypeif __name__ == '__main__':spark = SparkSession.builder.\appName("SparkSQL Example").\master("local[*]").\config("spark.sql.shuffle.partitions", "2").\config("spark.sql.warehouse.dir", "hdfs://node1:8020/user/hive/warehouse").\config("hive.metastore.uris", "thrift://node3:9083").\enableHiveSupport().\getOrCreate()df = spark.read.format("json").load("../../data/input/mini.json").\dropna(thresh=1, subset=['storeProvince']).\filter("storeProvince != 'null'").\filter("receivable < 10000").\select("storeProvince", "storeID", "receivable", "dateTS", "payType")province_sale_df = df.groupBy("storeProvince").sum("receivable").\withColumnRenamed("sum(receivable)", "money").\withColumn("money", F.round("money", 2)).\orderBy("money", ascending=False)province_sale_df.show(truncate=False)province_sale_df.write.mode("overwrite").\format("jdbc").\option("url", "jdbc:mysql://node1:3306/bigdata?useSSL=false&useUnicode=true&characterEncoding=utf8").\option("dbtable", "province_sale").\option("user", "root").\option("password", "123456").\option("encoding", "utf-8").\save()province_sale_df.write.mode("overwrite").saveAsTable("default.province_sale", "parquet")top3_province_df = province_sale_df.limit(3).select("storeProvince").withColumnRenamed("storeProvince", "top3_province")top3_province_df_joined = df.join(top3_province_df, on = df['storeProvince'] == top3_province_df['top3_province'])top3_province_df_joined.persist(StorageLevel.MEMORY_AND_DISK)province_hot_store_count_df = top3_province_df_joined.groupBy("storeProvince", "storeID",F.from_unixtime(df['dateTS'].substr(0,10), "yyyy-MM-dd").alias("day")).\sum("receivable").withColumnRenamed("sum(receivable)", "money").\filter("money > 1000").\dropDuplicates(subset=["storeID"]).\groupBy("storeProvince").count()province_hot_store_count_df.write.mode("overwrite").\format("jdbc").\option("url", "jdbc:mysql://node1:3306/bigdata?useSSL=false&useUnicode=true&characterEncoding=utf8").\option("dbtable", "province_hot_store_count").\option("user", "root").\option("password", "123456").\option("encoding", "utf-8").\save()province_hot_store_count_df.write.mode("overwrite").saveAsTable("default.province_hot_store_count", "parquet")top3_province_order_avg_df = top3_province_df_joined.groupBy("storeProvince").\avg("receivable").\withColumnRenamed("avg(receivable)", "money").\withColumn("money", F.round("money", 2)).\orderBy("money", ascending=False)top3_province_order_avg_df.show(truncate=False)top3_province_order_avg_df.write.mode("overwrite").\format("jdbc").\option("url", "jdbc:mysql://node1:3306/bigdata?useSSL=false&useUnicode=true&characterEncoding=utf8").\option("dbtable", "top3_province_order_avg").\option("user", "root").\option("password", "123456").\option("encoding", "utf-8").\save()top3_province_order_avg_df.write.mode("overwrite").saveAsTable("default.top3_province_order_avg", "parquet")top3_province_df_joined.createTempView("province_pay")def udf_func(percent):return str(round(percent * 100, 2)) + "%"my_udf = F.udf(udf_func, StringType())pay_type_df = spark.sql("""SELECT storeProvince, payType, (COUNT(payType) / total) AS percent FROM(SELECT storeProvince, payType, count(1) OVER(PARTITION BY storeProvince) AS total FROM province_pay) AS subGROUP BY storeProvince, payType, total""").withColumn("percent", my_udf("percent"))pay_type_df.show()pay_type_df.write.mode("overwrite").\format("jdbc").\option("url", "jdbc:mysql://node1:3306/bigdata?useSSL=false&useUnicode=true&characterEncoding=utf8").\option("dbtable", "pay_type").\option("user", "root").\option("password", "123456").\option("encoding", "utf-8").\save()pay_type_df.write.mode("overwrite").saveAsTable("default.pay_type", "parquet")top3_province_df_joined.unpersist()