当前位置： 首页 > article >正文

基于Spark 的零售交易数据挖掘分析与可视化

article 2024/11/16 5:52:58

基于Spark 的零售交易数据挖掘分析与可视化

本文将带你通过 PySpark 进行电商数据的分析处理，并将结果保存为 JSON 文件，供前端展示。我们将从数据的读取、处理、分析到结果保存和网页展示，覆盖完整的数据流。项目结构如下：

1、Spark 分析数据
2、生成 JSON 文件
3、使用 Bottle 框架搭建简单 Web 服务器

项目简介

我们使用了 PySpark 来处理一个电商数据集，数据存储在 HDFS 上。通过 SQL 和 RDD 操作实现了多个业务需求分析，并最终将结果保存为 JSON 文件，用于前端展示。后端 Web 服务采用 Bottle 框架，提供静态文件服务和页面展示。

数据集介绍

数据集包括了以下字段：

InvoiceNo: 订单号
StockCode: 商品编码
Description: 商品描述
Quantity: 数量
InvoiceDate: 订单日期
UnitPrice: 商品单价
CustomerID: 客户编号
Country: 国家
在这里插入图片描述

1. 数据读取

首先，我们从 HDFS 中读取 CSV 文件作为 Spark 的 DataFrame，并通过 createOrReplaceTempView 创建 SQL 查询视图。代码如下：

# 从HDFS中读取数据集为DataFrame
df = spark.read.format('com.databricks.spark.csv').options(header='true', inferschema='true').load('../data/E_Commerce_Data.csv')
df.createOrReplaceTempView("data")

2. 分析任务

通过 SQL 查询和 RDD 操作，项目实现了以下 10 项数据分析任务：

  1. 客户数最多的 10 个国家
    通过 SQL 查询，统计每个国家的客户数，并选出客户数最多的 10 个国家：
def countryCustomer():
    countryCustomerDF = spark.sql("SELECT Country,COUNT(DISTINCT CustomerID) AS countOfCustomer FROM data GROUP BY Country ORDER BY countOfCustomer DESC LIMIT 10")
    return countryCustomerDF.collect()

在这里插入图片描述

  1. 销量最高的 10 个国家
    统计每个国家的商品销量，并选出销量最高的 10 个国家：
def countryQuantity():
    countryQuantityDF = spark.sql("SELECT Country,SUM(Quantity) AS sumOfQuantity FROM data GROUP BY Country ORDER BY sumOfQuantity DESC LIMIT 10")
    return countryQuantityDF.collect()

在这里插入图片描述

  1. 各国总销售额分布
    计算每个国家的销售额，结果按销售额大小进行排序：
def countrySumOfPrice():
    countrySumOfPriceDF = spark.sql("SELECT Country,SUM(UnitPrice*Quantity) AS sumOfPrice FROM data GROUP BY Country")
    return countrySumOfPriceDF.collect()

在这里插入图片描述

  1. 销量最高的 10 个商品
    统计商品的销量，按销量大小选出销量最高的 10 个商品：
def stockQuantity():
    stockQuantityDF = spark.sql("SELECT StockCode,SUM(Quantity) AS sumOfQuantity FROM data GROUP BY StockCode ORDER BY sumOfQuantity DESC LIMIT 10")
    return stockQuantityDF.collect()

在这里插入图片描述

  1. 商品描述的热门关键词 Top 300
    通过对商品描述字段进行分词和词频统计，得到最热门的 300 个关键词：
def wordCount():
    wordCount = spark.sql("SELECT LOWER(Description) as description from data").rdd.filter(lambda line:line['description'] is not None).flatMap(lambda line:line['description'].split(" ")).map(lambda word:(word,1)).reduceByKey(lambda a,b:a+b).repartition(1).sortBy(lambda x:x[1],False)
    wordCountSchema = StructType([StructField("word", StringType(), True),StructField("count", IntegerType(), True)])
    wordCountDF = spark.createDataFrame(wordCount, wordCountSchema)
    return wordCountDF.take(300)

在这里插入图片描述

  1. 退货订单数最多的 10 个国家
    统计退货订单数量最多的 10 个国家，退货订单的 InvoiceNo 以 ‘C’ 开头：
def countryReturnInvoice():
    countryReturnInvoiceDF = spark.sql("SELECT Country,COUNT(DISTINCT InvoiceNo) AS countOfReturnInvoice FROM data WHERE InvoiceNo LIKE 'C%' GROUP BY Country ORDER BY countOfReturnInvoice DESC LIMIT 10")
    return countryReturnInvoiceDF.collect()

在这里插入图片描述

  1. 月销售额随时间的变化趋势
    通过提取 InvoiceDate 中的年份和月份，计算每月的销售额：
def tradePrice():
    result3 = formatData()
    result4 = result3.map(lambda line:(line[0]+"-"+line[1],line[3]*line[4]))
    result5 = result4.reduceByKey(lambda a,b:a+b).sortByKey()
    schema = StructType([StructField("date", StringType(), True),StructField("tradePrice", DoubleType(), True)])
    tradePriceDF = spark.createDataFrame(result5, schema)
    return tradePriceDF.collect()

在这里插入图片描述

  1. 日销量随时间的变化趋势
    计算每天的销售量变化趋势，提取 InvoiceDate 的年、月、日，并进行汇总：
def saleQuantity():
    result3 = formatData()
    result4 = result3.map(lambda line:(line[0]+"-"+line[1]+"-"+line[2],line[3]))
    result5 = result4.reduceByKey(lambda a,b:a+b).sortByKey()
    schema = StructType([StructField("date", StringType(), True),StructField("saleQuantity", IntegerType(), True)])
    saleQuantityDF = spark.createDataFrame(result5, schema)
    return saleQuantityDF.collect()

在这里插入图片描述

  1. 各国购买订单量与退货订单量的关系
    通过联表查询，展示每个国家的购买订单量与退货订单量的关系：
def buyReturn():
    returnDF = spark.sql("SELECT Country AS Country,COUNT(DISTINCT InvoiceNo) AS countOfReturn FROM data WHERE InvoiceNo LIKE 'C%' GROUP BY Country")
    buyDF = spark.sql("SELECT Country AS Country2,COUNT(DISTINCT InvoiceNo) AS countOfBuy FROM data WHERE InvoiceNo NOT LIKE 'C%' GROUP BY Country2")
    buyReturnDF = returnDF.join(buyDF, returnDF["Country"] == buyDF["Country2"], "left_outer")
    buyReturnDF = buyReturnDF.select(buyReturnDF["Country"],buyReturnDF["countOfBuy"],buyReturnDF["countOfReturn"])
    return buyReturnDF.collect()

在这里插入图片描述

  1. 商品的平均单价与销量的关系
    通过计算每个商品的平均单价和总销量，展示二者的关系：
def unitPriceSales():
    unitPriceSalesDF = spark.sql("SELECT StockCode,AVG(DISTINCT UnitPrice) AS avgUnitPrice,SUM(Quantity) AS sumOfQuantity FROM data GROUP BY StockCode")
    return unitPriceSalesDF.collect()

在这里插入图片描述

3. 数据结果保存

所有分析结果都以 JSON 格式保存到 static/ 目录。我们定义了一个简单的 save() 函数来处理文件写入：

def save(path, data):
    with open(path, 'w') as f:
        f.write(data)

4. 使用 Bottle 框架搭建 Web 服务器

为了展示这些分析结果，我们使用了 Bottle 框架，提供静态文件服务。Web 服务器代码如下：

from bottle import route, run, static_file

@route('/static/<filename>')
def server_static(filename):
    return static_file(filename, root="./static")

@route("/<name:re:.*\.html>")
def server_page(name):
    return static_file(name, root=".")

@route("/")
def index():
    return static_file("index.html", root=".")

通过访问 /static/filename 可以获取生成的 JSON 文件，访问 / 可以加载主页 index.html。

5. 运行项目

运行项目非常简单，只需启动 Python 脚本，它将自动生成分析结果，并启动 Bottle Web 服务器。

python app.py

在浏览器中访问 http://localhost:9999 即可查看分析结果。

在这里插入图片描述
在这里插入图片描述
在这里插入图片描述
在这里插入图片描述
在这里插入图片描述
在这里插入图片描述
在这里插入图片描述
在这里插入图片描述

总结

通过 PySpark 处理海量电商数据，并将结果可视化，是数据分析和数据工程领域的典型场景。本项目展示了如何通过 Spark 进行数据的处理和分析，结合 Bottle 框架实现简单的 Web 服务，将结果供用户查看。

如有遇到问题可以找小编沟通交流哦。另外小编帮忙辅导大课作业，学生毕设等。不限于MapReduce， MySQL, python，java，大数据，模型训练等。 hadoop hdfs yarn spark Django flask flink kafka flume datax sqoop seatunnel echart可视化 机器学习等
在这里插入图片描述


http://www.kler.cn/a/303204.html

相关文章：

  • LeetCode59. 螺旋矩阵 II
  • PyQt入门指南五十二 版本控制与协作开发
  • Ps：OpenColorIO 设置
  • 网络基础概念与应用：深入理解计算机网络
  • 【QT常用技术讲解】优化网络链接不上导致qt、qml界面卡顿的问题
  • NAT网络工作原理和NAT类型
  • Java | Leetcode Java题解之第403题青蛙过河
  • mysql一主2从部署
  • 记者协会评审系统-需求分析
  • 苹果CMS vs. 海洋CMS：哪个系统更易于百度收录？
  • 解决：Vue3 - defineProps 设置默认值报错问题
  • CentOS 入门基础
  • LeetCode 2398.预算内的最多机器人数目：滑动窗口+单调队列——思路清晰的一篇题解
  • 工业设备网关在智能工厂建设中发挥的作用，以某汽车制造厂为例
  • LabVIEW编程语言出于什么原因开发的?
  • 1688商品详情API返回值中的售后保障与服务信息
  • Failed building wheel for opencv-python-headless
  • k8s常用指令续：
  • Day 11-12：查找
  • day14-单例设计模式动态代理
  • Qt 学习第八天：菜单栏、工具栏、状态栏、模态和非模态对话框创建
  • RabbitMQ延迟消息——DelayExchange插件
  • Python之 条件与循环（Python‘s Conditions and loops）
  • 在麒麟系统 v10 SP3 上运行自带的 MariaDB
  • 【鸿蒙】HarmonyOS NEXT星河入门到实战6-组件化开发-样式结构重用常见组件
  • Oracle中VARCHAR和VARCHAR2的区别