• SparkSQL综合案例-省份维度的销售情况统计分析

    一、项目背景

    二、项目需求

            （1）需求

            ①各省销售指标，每个省份的销售额统计

            ②TOP3销售省份中，有多少家店铺日均销售额1000+

            ③TOP3省份中，各个省份的平均单价

            ④TOP3省份中，各个省份的支付类型比例

            （2）要求

            ①将需求结果写出到mysql

            ②将数据写入到Spark On Hive中

    三、代码实现

            （1）需求1：
    1. # cording:utf8
    2. '''
    3. 要求1：各省销售额统计
    4. 要求2：TOP3销售省份中，有多少店铺达到过日销售额1000+
    5. 要求3：TOP3省份中，各省的平均单单价
    6. 要求4：TOP3省份中，各个省份的支付类型比例
    7. '''
    8.  
    9. from pyspark.sql import SparkSession
    10. from pyspark.sql import functions as F
    11. if __name__ == '__main__':
    12.     spark = SparkSession.builder.appName('SQL_text').\
    13.         master('local[*]').\
    14.         config('spark.sql.shuffle.partitions', '2').\
    15.         config('spark.sql.warehouse.dir', 'hdfs://pyspark01/user/hive/warehouse').\
    16.         config('hive.metastore.uris', 'thrift://pyspark01:9083').\
    17.         enableHiveSupport().\
    18.         getOrCreate()
    19.     # 1.读取数据
    20.     # 省份信息，缺失值过滤，同时省份信息中会有‘null’字符串
    21.     # 订单的金额，数据集中有订单的金额是单笔超过10000的，这些事测试数据
    22.     # 列值过滤（SparkSQL会自动做这个优化）
    23.     df = spark.read.format('json').load('../../input/mini.json').\
    24.         dropna(thresh=1, subset=['storeProvince']).\
    25.         filter("storeProvince != 'null'").\
    26.         filter('receivable < 10000').\
    27.         select('storeProvince', 'storeID', 'receivable', 'dateTS', 'payType')
    28.  
    29.     # TODO 1:各省销售额统计
    30.     province_sale_df = df.groupBy('storeProvince').sum('receivable').\
    31.         withColumnRenamed('sum(receivable)', 'money').\
    32.         withColumn('money', F.round('money', 2)).\
    33.         orderBy('money', ascending=False)
    34.     province_sale_df.show(truncate=False)
    35.  
    36.     # 写出到Mysql
    37.     province_sale_df.write.mode('overwrite').\
    38.         format('jdbc').\
    39.         option('url', 'jdbc:mysql://pyspark01:3306/bigdata?useSSL=False&useUnicode=true&characterEncoding=utf8').\
    40.         option('dbtable', 'province_sale').\
    41.         option('user', 'root').\
    42.         option('password', 'root').\
    43.         option('encoding', 'utf8').\
    44.         save()
    45.  
    46.     # 写出Hive表 saveAsTable 可以写出表 要求已经配置好spark on hive
    47.     # 会将表写入到hive的数据仓库中
    48.     province_sale_df.write.mode('overwrite').saveAsTable('default.province_sale', 'parquet')

            结果展示：

            MySQL数据展示：

            Hive数据展示：

            （2）需求2：
    1. # cording:utf8
    2. '''
    3. 要求1：各省销售额统计
    4. 要求2：TOP3销售省份中，有多少店铺达到过日销售额1000+
    5. 要求3：TOP3省份中，各省的平均单单价
    6. 要求4：TOP3省份中，各个省份的支付类型比例
    7. '''
    8.  
    9. from pyspark.sql import SparkSession
    10. from pyspark.sql import functions as F
    11. from pyspark.storagelevel import StorageLevel
    12.  
    13. if __name__ == '__main__':
    14.     spark = SparkSession.builder.appName('SQL_text').\
    15.         master('local[*]').\
    16.         config('spark.sql.shuffle.partitions', '2').\
    17.         config('spark.sql.warehouse.dir', 'hdfs://pyspark01/user/hive/warehouse').\
    18.         config('hive.metastore.uris', 'thrift://pyspark01:9083').\
    19.         enableHiveSupport().\
    20.         getOrCreate()
    21.     # 1.读取数据
    22.     # 省份信息，缺失值过滤，同时省份信息中会有‘null’字符串
    23.     # 订单的金额，数据集中有订单的金额是单笔超过10000的，这些事测试数据
    24.     # 列值过滤（SparkSQL会自动做这个优化）
    25.     df = spark.read.format('json').load('../../input/mini.json').\
    26.         dropna(thresh=1, subset=['storeProvince']).\
    27.         filter("storeProvince != 'null'").\
    28.         filter('receivable < 10000').\
    29.         select('storeProvince', 'storeID', 'receivable', 'dateTS', 'payType')
    30.  
    31.     # TODO 1:各省销售额统计
    32.     province_sale_df = df.groupBy('storeProvince').sum('receivable').\
    33.         withColumnRenamed('sum(receivable)', 'money').\
    34.         withColumn('money', F.round('money', 2)).\
    35.         orderBy('money', ascending=False)
    36.     # # 写出到Mysql
    37.     # province_sale_df.write.mode('overwrite').\
    38.     #     format('jdbc').\
    39.     #     option('url', 'jdbc:mysql://pyspark01:3306/bigdata?useSSL=False&useUnicode=true&characterEncoding=utf8').\
    40.     #     option('dbtable', 'province_sale').\
    41.     #     option('user', 'root').\
    42.     #     option('password', 'root').\
    43.     #     option('encoding', 'utf8').\
    44.     #     save()
    45.     #
    46.     # # 写出Hive表 saveAsTable 可以写出表 要求已经配置好spark on hive
    47.     # # 会将表写入到hive的数据仓库中
    48.     # province_sale_df.write.mode('overwrite').saveAsTable('default.province_sale', 'parquet')
    49.  
    50.     # TODO 需求2：TOP3销售省份中，有多少店铺达到过日销售额1000+
    51.     # 2.1 找到TOP3的销售省份
    52.     top3_province_df = province_sale_df.limit(3).select('storeProvince').\
    53.         withColumnRenamed('storeProvince', 'top3_province')     # 对列名进行重命名，防止与province_sale_df的storeProvince冲突
    54.  
    55.     # 2.2 和原始的DF进行内关联，数据关联后，得到TOP3省份的销售数据
    56.     top3_province_df_joined = df.join(top3_province_df, on=df['storeProvince'] == top3_province_df['top3_province'])
    57.  
    58.     # 因为需要多次使用到TOP3省份数据，所有对其进行持久化缓存
    59.     top3_province_df_joined.persist(StorageLevel.MEMORY_AND_DISK)
    60.  
    61.     # from_unixtime将秒级的日期数据转换为年月日数据
    62.     # from_unixtime的精度是秒级，数据的精度是毫秒级，需要对数据进行进度的裁剪
    63.     province_hot_store_count_df = top3_province_df_joined.groupBy("storeProvince", "storeID",
    64.                         F.from_unixtime(df['dateTS'].substr(0, 10), "yyyy-mm-dd").alias('day')).\
    65.                         sum('receivable').withColumnRenamed('sum(receivable)', 'money').\
    66.                          filter('money > 1000 ').\
    67.                         dropDuplicates(subset=['storeID']).\
    68.                         groupBy('storeProvince').count()
    69.     province_hot_store_count_df.show()
    70.     # 写出到Mysql
    71.     province_sale_df.write.mode('overwrite'). \
    72.         format('jdbc'). \
    73.         option('url', 'jdbc:mysql://pyspark01:3306/bigdata?useSSL=False&useUnicode=true&characterEncoding=utf8'). \
    74.         option('dbtable', 'province_hot_store_count'). \
    75.         option('user', 'root'). \
    76.         option('password', 'root'). \
    77.         option('encoding', 'utf8'). \
    78.         save()
    79.  
    80.     # 写出Hive表 saveAsTable 可以写出表 要求已经配置好spark on hive
    81.     # 会将表写入到hive的数据仓库中
    82.     province_sale_df.write.mode('overwrite').saveAsTable('default.province_hot_store_count', 'parquet')
    83.  
    84.     top3_province_df_joined.unpersist()

            结果展示：

            MySQL结果展示：

            Hive结果展示：

            （3）需求3：
    1. # cording:utf8
    2. '''
    3. 要求1：各省销售额统计
    4. 要求2：TOP3销售省份中，有多少店铺达到过日销售额1000+
    5. 要求3：TOP3省份中，各省的平均单单价
    6. 要求4：TOP3省份中，各个省份的支付类型比例
    7. '''
    8.  
    9. from pyspark.sql import SparkSession
    10. from pyspark.sql import functions as F
    11. from pyspark.storagelevel import StorageLevel
    12.  
    13. if __name__ == '__main__':
    14.     spark = SparkSession.builder.appName('SQL_text').\
    15.         master('local[*]').\
    16.         config('spark.sql.shuffle.partitions', '2').\
    17.         config('spark.sql.warehouse.dir', 'hdfs://pyspark01/user/hive/warehouse').\
    18.         config('hive.metastore.uris', 'thrift://pyspark01:9083').\
    19.         enableHiveSupport().\
    20.         getOrCreate()
    21.     # 1.读取数据
    22.     # 省份信息，缺失值过滤，同时省份信息中会有‘null’字符串
    23.     # 订单的金额，数据集中有订单的金额是单笔超过10000的，这些事测试数据
    24.     # 列值过滤（SparkSQL会自动做这个优化）
    25.     df = spark.read.format('json').load('../../input/mini.json').\
    26.         dropna(thresh=1, subset=['storeProvince']).\
    27.         filter("storeProvince != 'null'").\
    28.         filter('receivable < 10000').\
    29.         select('storeProvince', 'storeID', 'receivable', 'dateTS', 'payType')
    30.  
    31.     # TODO 1:各省销售额统计
    32.     province_sale_df = df.groupBy('storeProvince').sum('receivable').\
    33.         withColumnRenamed('sum(receivable)', 'money').\
    34.         withColumn('money', F.round('money', 2)).\
    35.         orderBy('money', ascending=False)
    36.     # # 写出到Mysql
    37.     # province_sale_df.write.mode('overwrite').\
    38.     #     format('jdbc').\
    39.     #     option('url', 'jdbc:mysql://pyspark01:3306/bigdata?useSSL=False&useUnicode=true&characterEncoding=utf8').\
    40.     #     option('dbtable', 'province_sale').\
    41.     #     option('user', 'root').\
    42.     #     option('password', 'root').\
    43.     #     option('encoding', 'utf8').\
    44.     #     save()
    45.     #
    46.     # # 写出Hive表 saveAsTable 可以写出表 要求已经配置好spark on hive
    47.     # # 会将表写入到hive的数据仓库中
    48.     # province_sale_df.write.mode('overwrite').saveAsTable('default.province_sale', 'parquet')
    49.  
    50.     # TODO 需求2：TOP3销售省份中，有多少店铺达到过日销售额1000+
    51.     # 2.1 找到TOP3的销售省份
    52.     top3_province_df = province_sale_df.limit(3).select('storeProvince').\
    53.         withColumnRenamed('storeProvince', 'top3_province')     # 对列名进行重命名，防止与province_sale_df的storeProvince冲突
    54.  
    55.     # 2.2 和原始的DF进行内关联，数据关联后，得到TOP3省份的销售数据
    56.     top3_province_df_joined = df.join(top3_province_df, on=df['storeProvince'] == top3_province_df['top3_province'])
    57.  
    58.     # 因为需要多次使用到TOP3省份数据，所有对其进行持久化缓存
    59.     top3_province_df_joined.persist(StorageLevel.MEMORY_AND_DISK)
    60.  
    61.     # from_unixtime将秒级的日期数据转换为年月日数据
    62.     # from_unixtime的精度是秒级，数据的精度是毫秒级，需要对数据进行进度的裁剪
    63.     province_hot_store_count_df = top3_province_df_joined.groupBy("storeProvince", "storeID",
    64.                         F.from_unixtime(df['dateTS'].substr(0, 10), "yyyy-mm-dd").alias('day')).\
    65.                         sum('receivable').withColumnRenamed('sum(receivable)', 'money').\
    66.                          filter('money > 1000 ').\
    67.                         dropDuplicates(subset=['storeID']).\
    68.                         groupBy('storeProvince').count()
    69.     province_hot_store_count_df.show()
    70.     # # 写出到Mysql
    71.     # province_hot_store_count_df.write.mode('overwrite'). \
    72.     #     format('jdbc'). \
    73.     #     option('url', 'jdbc:mysql://pyspark01:3306/bigdata?useSSL=False&useUnicode=true&characterEncoding=utf8'). \
    74.     #     option('dbtable', 'province_hot_store_count'). \
    75.     #     option('user', 'root'). \
    76.     #     option('password', 'root'). \
    77.     #     option('encoding', 'utf8'). \
    78.     #     save()
    79.     #
    80.     # # 写出Hive表 saveAsTable 可以写出表 要求已经配置好spark on hive
    81.     # # 会将表写入到hive的数据仓库中
    82.     # province_hot_store_count_df.write.mode('overwrite').saveAsTable('default.province_hot_store_count', 'parquet')
    83.  
    84.  
    85.     # TODO 3:TOP3省份中，各省的平均单单价
    86.     top3_province_order_avg_df = top3_province_df_joined.groupBy("storeProvince").\
    87.         avg("receivable").\
    88.         withColumnRenamed("avg(receivable)", "money").\
    89.         withColumn("money", F.round("money", 2)).\
    90.         orderBy("money", ascending=False)
    91.     top3_province_order_avg_df.show(truncate=False)
    92.  
    93.     # 写出到Mysql
    94.     top3_province_order_avg_df.write.mode('overwrite'). \
    95.         format('jdbc'). \
    96.         option('url', 'jdbc:mysql://pyspark01:3306/bigdata?useSSL=False&useUnicode=true&characterEncoding=utf8'). \
    97.         option('dbtable', 'top3_province_order_avg'). \
    98.         option('user', 'root'). \
    99.         option('password', 'root'). \
    100.         option('encoding', 'utf8'). \
    101.         save()
    102.  
    103.     # 写出Hive表 saveAsTable 可以写出表 要求已经配置好spark on hive
    104.     # 会将表写入到hive的数据仓库中
    105.     top3_province_order_avg_df.write.mode('overwrite').saveAsTable('default.top3_province_order_avg', 'parquet')
    106.  
    107.     top3_province_df_joined.unpersist()

            结果展示

            MySQL与Hive结果展示：

            （4）需求4：
    1. # cording:utf8
    2. '''
    3. 要求1：各省销售额统计
    4. 要求2：TOP3销售省份中，有多少店铺达到过日销售额1000+
    5. 要求3：TOP3省份中，各省的平均单单价
    6. 要求4：TOP3省份中，各个省份的支付类型比例
    7. '''
    8.  
    9. from pyspark.sql import SparkSession
    10. from pyspark.sql import functions as F
    11. from pyspark.storagelevel import StorageLevel
    12. from pyspark.sql.types import StringType
    13. if __name__ == '__main__':
    14.     spark = SparkSession.builder.appName('SQL_text').\
    15.         master('local[*]').\
    16.         config('spark.sql.shuffle.partitions', '2').\
    17.         config('spark.sql.warehouse.dir', 'hdfs://pyspark01/user/hive/warehouse').\
    18.         config('hive.metastore.uris', 'thrift://pyspark01:9083').\
    19.         enableHiveSupport().\
    20.         getOrCreate()
    21.     # 1.读取数据
    22.     # 省份信息，缺失值过滤，同时省份信息中会有‘null’字符串
    23.     # 订单的金额，数据集中有订单的金额是单笔超过10000的，这些事测试数据
    24.     # 列值过滤（SparkSQL会自动做这个优化）
    25.     df = spark.read.format('json').load('../../input/mini.json').\
    26.         dropna(thresh=1, subset=['storeProvince']).\
    27.         filter("storeProvince != 'null'").\
    28.         filter('receivable < 10000').\
    29.         select('storeProvince', 'storeID', 'receivable', 'dateTS', 'payType')
    30.  
    31.     # TODO 1:各省销售额统计
    32.     province_sale_df = df.groupBy('storeProvince').sum('receivable').\
    33.         withColumnRenamed('sum(receivable)', 'money').\
    34.         withColumn('money', F.round('money', 2)).\
    35.         orderBy('money', ascending=False)
    36.     # # 写出到Mysql
    37.     # province_sale_df.write.mode('overwrite').\
    38.     #     format('jdbc').\
    39.     #     option('url', 'jdbc:mysql://pyspark01:3306/bigdata?useSSL=False&useUnicode=true&characterEncoding=utf8').\
    40.     #     option('dbtable', 'province_sale').\
    41.     #     option('user', 'root').\
    42.     #     option('password', 'root').\
    43.     #     option('encoding', 'utf8').\
    44.     #     save()
    45.     #
    46.     # # 写出Hive表 saveAsTable 可以写出表 要求已经配置好spark on hive
    47.     # # 会将表写入到hive的数据仓库中
    48.     # province_sale_df.write.mode('overwrite').saveAsTable('default.province_sale', 'parquet')
    49.  
    50.     # TODO 需求2：TOP3销售省份中，有多少店铺达到过日销售额1000+
    51.     # 2.1 找到TOP3的销售省份
    52.     top3_province_df = province_sale_df.limit(3).select('storeProvince').\
    53.         withColumnRenamed('storeProvince', 'top3_province')     # 对列名进行重命名，防止与province_sale_df的storeProvince冲突
    54.  
    55.     # 2.2 和原始的DF进行内关联，数据关联后，得到TOP3省份的销售数据
    56.     top3_province_df_joined = df.join(top3_province_df, on=df['storeProvince'] == top3_province_df['top3_province'])
    57.  
    58.     # 因为需要多次使用到TOP3省份数据，所有对其进行持久化缓存
    59.     top3_province_df_joined.persist(StorageLevel.MEMORY_AND_DISK)
    60.  
    61.     # from_unixtime将秒级的日期数据转换为年月日数据
    62.     # from_unixtime的精度是秒级，数据的精度是毫秒级，需要对数据进行进度的裁剪
    63.     province_hot_store_count_df = top3_province_df_joined.groupBy("storeProvince", "storeID",
    64.                         F.from_unixtime(df['dateTS'].substr(0, 10), "yyyy-mm-dd").alias('day')).\
    65.                         sum('receivable').withColumnRenamed('sum(receivable)', 'money').\
    66.                          filter('money > 1000 ').\
    67.                         dropDuplicates(subset=['storeID']).\
    68.                         groupBy('storeProvince').count()
    69.     province_hot_store_count_df.show()
    70.     # # 写出到Mysql
    71.     # province_hot_store_count_df.write.mode('overwrite'). \
    72.     #     format('jdbc'). \
    73.     #     option('url', 'jdbc:mysql://pyspark01:3306/bigdata?useSSL=False&useUnicode=true&characterEncoding=utf8'). \
    74.     #     option('dbtable', 'province_hot_store_count'). \
    75.     #     option('user', 'root'). \
    76.     #     option('password', 'root'). \
    77.     #     option('encoding', 'utf8'). \
    78.     #     save()
    79.     #
    80.     # # 写出Hive表 saveAsTable 可以写出表 要求已经配置好spark on hive
    81.     # # 会将表写入到hive的数据仓库中
    82.     # province_hot_store_count_df.write.mode('overwrite').saveAsTable('default.province_hot_store_count', 'parquet')
    83.  
    84.  
    85.     # TODO 3:TOP3省份中，各省的平均单单价
    86.     top3_province_order_avg_df = top3_province_df_joined.groupBy("storeProvince").\
    87.         avg("receivable").\
    88.         withColumnRenamed("avg(receivable)", "money").\
    89.         withColumn("money", F.round("money", 2)).\
    90.         orderBy("money", ascending=False)
    91.     top3_province_order_avg_df.show(truncate=False)
    92.  
    93.     # # 写出到Mysql
    94.     # top3_province_order_avg_df.write.mode('overwrite'). \
    95.     #     format('jdbc'). \
    96.     #     option('url', 'jdbc:mysql://pyspark01:3306/bigdata?useSSL=False&useUnicode=true&characterEncoding=utf8'). \
    97.     #     option('dbtable', 'top3_province_order_avg'). \
    98.     #     option('user', 'root'). \
    99.     #     option('password', 'root'). \
    100.     #     option('encoding', 'utf8'). \
    101.     #     save()
    102.     #
    103.     # # 写出Hive表 saveAsTable 可以写出表 要求已经配置好spark on hive
    104.     # # 会将表写入到hive的数据仓库中
    105.     # top3_province_order_avg_df.write.mode('overwrite').saveAsTable('default.top3_province_order_avg', 'parquet')
    106.  
    107.     # TODO 4：TOP3省份中，各个省份的支付类型比例
    108.     top3_province_df_joined.createTempView("province_pay")
    109.  
    110.     # 自定义UDF
    111.     def udf_func(percent):
    112.         return str(round(percent * 100)) + "%"
    113.     # 注册UDF
    114.     my_udf = F.udf(udf_func, StringType())
    115.  
    116.     pay_type_df = spark.sql('''
    117.         SELECT storeProvince, payType, (count(payType) / total) AS percent FROM 
    118.         (SELECT storeProvince, payType, count(1) OVER(PARTITION BY storeProvince) AS total FROM province_pay) AS sub
    119.         GROUP BY storeProvince, payType, total
    120.     ''').withColumn('percent', my_udf("percent"))
    121.     pay_type_df.show()
    122.  
    123.     # 写出到Mysql
    124.     pay_type_df.write.mode('overwrite'). \
    125.         format('jdbc'). \
    126.         option('url', 'jdbc:mysql://pyspark01:3306/bigdata?useSSL=False&useUnicode=true&characterEncoding=utf8'). \
    127.         option('dbtable', 'pay_type'). \
    128.         option('user', 'root'). \
    129.         option('password', 'root'). \
    130.         option('encoding', 'utf8'). \
    131.         save()
    132.  
    133.     # 写出Hive表 saveAsTable 可以写出表 要求已经配置好spark on hive
    134.     # 会将表写入到hive的数据仓库中
    135.     top3_province_order_avg_df.write.mode('overwrite').saveAsTable('default.pay_type', 'parquet')
    136.  
    137.     top3_province_df_joined.unpersist()

           结果展示：

           MySQL结果展示：

            Hive结果展示：

    四、项目运行问题及解决方法

            报错：java.sql.BatchUpdateException: Incorrect string value: '\xE6\xB1\x9F\xE8\xA5\xBF...' for column 'storeProvince' atrow1

            原因：MySQL的UTF-8只支持3个字节的unicode字符，无法支持四个字节的Unicode字符

            解决办法：在MySQL控制台执行下列代码修改编码格式

  • 相关阅读:
    测试工作流程图，你一定要知道的
    【利用Selenium+autoIt实现文件上传】
    好用的读书网站
    Java集合（二）：Map集合与Collections工具类
    Spring Cloud Alibaba（四）
    DJ12-2-2 算术运算指令
    OpenCV入门5——OpenCV的算术与位运算
    水管安装过滤器笔记
    C语言——结构体（位段）、联合体、枚举
    论文阅读 Dynamic Network Embedding by Modeling Triadic Closure Process
  • 原文地址：https://blog.csdn.net/2202_75347029/article/details/134083558