0
0

大数据实战（hadoop+spark+python）：淘宝电商数据分析

一，运行环境与所需资源：

  1. 虚拟机：Ubuntu 20.04.6 LTS

  2. docker容器

  3. hadoop-3.3.4

  4. spark-3.3.2-bin-hadoop3

  5. python，pyspark, pandas，matplotlib

  6. mysql，mysql-connector-j-8.0.32.jar（下载不需要积分什么的）

  7. 淘宝用户数据

  8. 以上的技术积累需要自行完成

二，创建与配置分布式镜像

  1. 创建主节点

  • 创建容器（##ubuntu的代码块，在ubuntu中运行，无特殊说明的在docker中运行）

  1. ##ubuntu
  2. #创建挂载目录
  3. sudo mkdir bigdata
  4. #拉取镜像
  5. docker pull ubuntu
  6. #创建容器
  7. docker run -it --name master --privileged=true -p 9870:9870 -p 8088:8088 -p 8080:8080 -p 7077:7077 -p 18080:18080 -p 8032:8032 --network hadoopnet --ip 172.20.0.5 -v /bigdata:/bigdata ubuntu /bin/bash

  • 更新软件列表，安装java，ssh，pip，pyspark，pyarrow, vim，mysql，pandas，zip，matplotlib

代码
  1. ##ubuntu
  2. #将mysql-connector-j-8.0.32.jar上传到挂载卷
  3. sudo mv mysql-connector-j-8.0.32.jar /bigdata
  1. apt update
  2. apt install vim
  3. apt insatll zip
  4. apt install pip
  5. pip install pyspark
  6. pip install pyarrow
  7. pip install pandas
  8. pip install mmatplotlib
  9. apt install openjdk-8-jdk
  10. #查看安装结果
  11. java -version
  12. #安装ssh，生成秘钥，并加入授权文件，启动ssh
  13. apt install openssh-server
  14. apt install openssh-client
  15. ssh-keygen -t rsa -P ""
  16. cd ~/.ssh
  17. cat id_rsa.pub >> authorized_keys
  18. service ssh start
  19. #查看安装结果
  20. ssh localhost
  21. #安装mysql，配置密码，第一次进入无密码，一直enter就行
  22. apt install mysql-server
  23. service mysql start
  24. mysql -u root -p
  25. #设置密码（命令在mysql里面运行），密码是：mynewpassword
  26. ALTER USER 'root'@'localhost' IDENTIFIED WITH mysql_native_password by 'mynewpassword';
  27. quit;
  28. #配置安全措施（docker容器运行）
  29. mysql_secure_installation
  31. #由于链接mysql需要mysql-connector-j-8.0.32.jar
  32. #所以要下载这个jar，然后添加到pyspark库的安装包下，和$JAVA_HOME的jre/lib/ext下
  33. #首先需要找到python运行环境，由于没有配置spark的python运行环境，默认ubuntu的python运行环境
  34. #就需要通过命令find /usr -name python 找到python文件夹，位于/usr/lib下
  35. find /usr -name python
  36. cd /usr/local/lib/python3.10/dist-packages
  37. cp /bigdata/mysql-connector-j-8.0.32.jar ./pyspark
  38. cp /bigdata/mysql-connector-j-8.0.32.jar $JAVA_HOME/jre/lib/ext
运行结果
  1. #java -version，运行结果
  2. openjdk version "1.8.0_362"
  3. OpenJDK Runtime Environment (build 1.8.0_362-8u362-ga-0ubuntu1~22.04-b09)
  4. OpenJDK 64-Bit Server VM (build 25.362-b09, mixed mode)
  6. #ssh localhost，运行结果，需要按一下y
  7. ssh: Could not resolve hostname hostlocal: Temporary failure in name resolution
  8. root@355a1f302b29:~/.ssh# ssh localhost 
  9. The authenticity of host 'localhost (127.0.0.1)' can't be established.
  10. ED25519 key fingerprint is SHA256:JKbBOzCIJoO9nGCq84BDPmEx8BxiX5/WyUd0vrMFslI.
  11. This key is not known by any other names
  12. Are you sure you want to continue connecting (yes/no/[fingerprint])? yes
  13. Warning: Permanently added 'localhost' (ED25519) to the list of known hosts.
  14. Welcome to Ubuntu 22.04.2 LTS (GNU/Linux 5.15.0-67-generic x86_64)
  16.  * Documentation:  https://help.ubuntu.com
  17.  * Management:     https://landscape.canonical.com
  18.  * Support:        https://ubuntu.com/advantage
  20. This system has been minimized by removing packages and content that are
  21. not required on a system that users do not log into.
  23. To restore this content, you can run the 'unminimize' command.
  25. The programs included with the Ubuntu system are free software;
  26. the exact distribution terms for each program are described in the
  27. individual files in /usr/share/doc/*/copyright.
  29. Ubuntu comes with ABSOLUTELY NO WARRANTY, to the extent permitted by
  30. applicable law.
  32. ##mysql_secure_installation，运行内容
  33. #第一次y/n，设置root密码
  34. VALIDATE PASSWORD COMPONENT can be used to test passwords
  35. and improve security. It checks the strength of password
  36. and allows the users to set only those passwords which are
  37. secure enough. Would you like to setup VALIDATE PASSWORD component?
  38. Press y|Y for Yes, any other key for No: y
  40. #设置密码强度
  41. There are three levels of password validation policy:
  43. LOW    Length >= 8
  44. MEDIUM Length >= 8, numeric, mixed case, and special characters
  45. STRONG Length >= 8, numeric, mixed case, special characters and dictionary                  file
  47. Please enter 0 = LOW, 1 = MEDIUM and 2 = STRONG: 0
  49. #输入密码
  50. New password: 
  52. Re-enter new password: 
  54. #是否删除隐秘用户
  55. Remove anonymous users? (Press y|Y for Yes, any other key for No) : y
  57. #root用户远程登录
  58. Disallow root login remotely? (Press y|Y for Yes, any other key for No) : y
  60. #是否删除test数据库
  61. Remove test database and access to it? (Press y|Y for Yes, any other key for No) : n
  63. #提交配置，刷新
  64. Reload privilege tables now? (Press y|Y for Yes, any other key for No) : y
  66. #find /usr -name python运行结果
  67. /usr/local/lib/python3.10/dist-packages/pyspark/python
  68. /usr/local/lib/python3.10/dist-packages/pyspark/examples/src/main/python
  69. /usr/local/spark/kubernetes/dockerfiles/spark/bindings/python
  70. /usr/local/spark/python
  71. /usr/local/spark/python/pyspark/python
  72. /usr/local/spark/examples/src/main/python
  73. /usr/share/gcc/python

  • 安装hadoop，spark

  1. ##ubuntu
  2. #上传压缩包到挂载目录，压缩安装包要自行下载
  3. sudo mv hadoop-3.3.4.tar.gz /bigdata
  4. sudo mv spark-3.3.2-bin-hadoop3.tgz /bigdata
  1. #查看挂载卷有没有安装包
  2. ls /bigdata
  3. #安装hadoop，spark
  4. tar -zxvf /bigdata/hadoop-3.3.4.tar.gz -C /usr/local/
  5. tar -zxvf /bigdata/spark-3.3.2-bin-hadoop3.tgz -C /usr/local/
  6. #更改目录名称方便使用
  7. cd /usr/local
  8. mv hadoop-3.3.4 hadoop
  9. mv spark-3.3.2-bin-hadoop3 spark

  • 配置环境

  1. #vim内容在下面
  2. vim ~/.bashrc
  3. source ~/.bashrc
  4. cd /usr/local/hadoop/etc/hadoop
  5. vim hadoop-env.sh
  6. vim hdfs-site.xml
  7. vim core-site.xml
  8. vim yarn-site.xml
  9. vim mapred-site.xml
  10. vim workers
  11. cd /usr/local/spark/conf
  12. mv spark-env.sh.template spark-env.sh
  13. vim spark-env.sh
  14. mv workers.template workers
  15. vim workers
  16. #格式化
  17. cd /usr/local/hadoop
  18. ./bin/hdfs -namenode
  20. #查看结果
  21. echo $JAVA_HOME
.bashrc
  1. #java，路径看自己本机
  2. export JAVA_HOME=/usr/lib/jvm/java-8-openjdk-amd64
  3. export JRE_HOME=${JAVA_HOME}/jre
  4. export CLASSPATH=.:${JAVA_HOME}/lib:${JRE_HOME}/lib
  5. export PATH=${JAVA_HOME}/bin:$PATH
  7. #hadoop，路径要看自己本机
  8. export HADOOP_HOME=/usr/local/hadoop
  10. #spark,具体路径看自己本机
  11. export SPARK_HOME=/usr/local/spark
  13. export PATH=$HADOOP_HOME/bin:$HADOOP_HOME/sbin:$JAVA_HOME/bin:$SPARK_HOME/bin:$PATH
hadoop-env.sh
  1. #具体路径看自己本机，HADOOP_PID_DIR指定hadoop进程号存放目录，最好不要放在ubuntu中的/tmp
  2. #因为ubuntu会隔一段时间清一下/tmp，会杀死hadoop
  3. export JAVA_HOME=/usr/lib/jvm/java-8-openjdk-amd64
  4. export HDFS_NAMENODE_USER=root
  5. export HDFS_DATANODE_USER=root
  6. export HDFS_SECONDARYNAMENODE_USER=root
  7. export YARN_RESOURCEMANAGER_USER=root
  8. export YARN_NODEMANAGER_USER=root
  9. export HADOOP_PID_DIR=/usr/local/hadoop/tmp
hdfs-site.xml
  1. <property>
  2.         <name>dfs.replication</name>
  3.         <value>2</value>
  4.     </property>
  5.     <property>
  6.         <name>dfs.namenode.name.dir</name>
  7.         <value>file:/usr/local/hadoop/tmp/dfs/name</value>
  8.     </property>
  9.     <property>
  10.         <name>dfs.datanode.data.dir</name>
  11.         <value>file:/usr/local/hadoop/tmp/dfs/data</value>
  12.   </property>
core-site.xml
  1. #与伪分布式不同，172.20.0.5，为master的IP地址
  2. <property>
  3.         <name>hadoop.tmp.dir</name>
  4.         <value>file:/usr/local/hadoop/tmp</value>
  5.         <description>Abase for other temporary directories.</description>
  6.     </property>
  7.     <property>
  8.         <name>fs.defaultFS</name>
  9.         <value>hdfs://172.20.0.5:9000</value>
  10. </property>
yarn-site.xml
  1. #yarn.nodemanager.aux-services：用于支持MapReduce应用程序的本地化任务执行和数据缓存
  2. #yarn.resourcemanager.address：设置resourcemanager运行地址，不设置，spark会找不到
  3. <property>
  4.         <name>yarn.resourcemanager.hostname</name>
  5.         <value>172.20.0.5</value>
  6. </property>
  7. <property>
  8.         <name>yarn.nodemanager.aux-services</name>
  9.         <value>mapreduce_shuffle</value>
  10. </property>
  12.  <property>
  13.         <name>mapreduce.application.classpath</name>
  14.         <value>
  15.             /usr/local/hadoop/etc/hadoop,
  16.             /usr/local/hadoop/share/hadoop/common/*,
  17.             /usr/local/hadoop/share/hadoop/common/lib/*,
  18.             /usr/local/hadoop/share/hadoop/hdfs/*,
  19.             /usr/local/hadoop/share/hadoop/hdfs/lib/*,
  20.             /usr/local/hadoop/share/hadoop/mapreduce/*,
  21.             /usr/local/hadoop/share/hadoop/mapreduce/lib/*,
  22.             /usr/local/hadoop/share/hadoop/yarn/*,
  23.             /usr/local/hadoop/share/hadoop/yarn/lib/*
  24.         </value>
  25.     </property>
  26. <property>
  27.         <name>yarn.resourcemanager.address</name>
  28.         <value>172.20.0.5:8032</value>
  29. </property>
mapred-site.xml
  1. #在yarn上运行mapreduce任务
  2. <property>
  3.         <name>mapreduce.framework.name</name>
  4.         <value>yarn</value>
  5. </property>
workers
  1. #hadoop的worker，位于/usr/local/hadoop/etc/hadoop下
  2. #数据节点IP地址，后面会创建
  3. 172.20.0.5
  4. 172.20.0.6
  5. 172.20.0.7
  6. 172.20.0.8
spark-env.sh
  1. #告诉HADOOP_CONF_DIR，spark会自己了解hadoop的环境配置，比如hdfs开放端口在哪
  2. #
  3. export HADOOP_CONF_DIR=/usr/local/hadoop/etc/hadoop
  4. export LD_LIBRARY_PATH=$HADOOP_HOME/lib/native
workers
  1. #spark的worker，位于/usr/local/spark/conf下
  2. #节点的IP地址，后面会创建
  3. 172.20.0.5
  4. 172.20.0.6
  5. 172.20.0.7
  6. 172.20.0.8
运行结果
  1. #格式化运行结果
  2. /************************************************************
  3. SHUTDOWN_MSG: Shutting down NameNode at 355a1f302b29/172.20.0.5
  4. ************************************************************/
  6. #echo $JAVA_HOME
  7. /usr/lib/jvm/java-8-openjdk-amd64

  • 保存镜像

  1. ##ubuntu
  2. docker commit master spark_images:v1

  1. 创建工作节点，构建分布式hadoop和spark

  • 代码
  1. ##ubuntu
  2. docker run -itd --name worker1 --net hadoopnet --ip 172.20.0.6 -v /bigdata:/bigdata spark_images:v1 /bin/bash
  3. docker run -itd --name worker2 --net hadoopnet --ip 172.20.0.7 -v /bigdata:/bigdata spark_images:v1 /bin/bash
  4. docker run -itd --name worker3 --net hadoopnet --ip 172.20.0.8 -v /bigdata:/bigdata spark_images:v1 /bin/bash
  1. #在每一个容器里都开启ssh，然后都互联一次，例如worker1，互联前要确保每个节点都开启了ssh
  2. service ssh start
  3. ssh 172.20.0.5
  4. ssh 172.20.0.7
  5. ssh 172.20.0.8
  7. #全部互联完后，回到master节点，开启hadoop，spark，mysql
  8. #然后回到ubuntu，用浏览器登录localhost:9870，localhost:8080,查看hdfs，spark是否可以访问
  9. service mysql start
  10. cd /usr/local/hadoop
  11. ./sbin/start-all.sh
  12. cd /usr/local/spark
  13. ./sbin/start-all.sh
  14. jps
  16. #运行示例1
  17. cd /usr/local/hadoop
  18. cp ./etc/hadoop/core-site.xml input
  19. ./bin/hdfs dfs -mkdir -p /user/root
  20. ./bin/hdfs dfs -put ./input .
  21. ./bin/hadoop jar ./share/hadoop/mapreduce/hadoop-mapreduce-examples-3.3.4.jar wordcount input output
  23. #运行示例2
  24. ./bin/spark-submit --class org.apache.spark.examples.SparkPi \
  25. --master yarn \
  26. --deploy-mode cluster \
  27. --driver-memory 2g \
  28. --executor-memory 1g \
  29. --executor-cores 2 \
  30. examples/jars/spark-examples*.jar \
  31. 10
  • 运行结果
  1. #jps
  2. 2628 Jps
  3. 2118 NodeManager
  4. 2008 ResourceManager
  5. 2570 Worker
  6. 1772 SecondaryNameNode
  7. 1484 NameNode
  8. 2494 Master
  9. 1599 DataNode
  11. #示例1运行结果
  12. Shuffle Errors
  13.         BAD_ID=0
  14.         CONNECTION=0
  15.         IO_ERROR=0
  16.         WRONG_LENGTH=0
  17.         WRONG_MAP=0
  18.         WRONG_REDUCE=0
  19.     File Input Format Counters 
  20.         Bytes Read=1068
  21.     File Output Format Counters 
  22.         Bytes Written=1039
  24. #示例2运行结果
  25.      client token: N/A
  26.      diagnostics: N/A
  27.      ApplicationMaster host: 0a69e22f207a
  28.      ApplicationMaster RPC port: 41551
  29.      queue: default
  30.      start time: 1679407401940
  31.      final status: SUCCEEDED
  32.      tracking URL: http://355a1f302b29:8088/proxy/application_1679405990349_0003/
  33.      user: root
  34. 23/03/21 14:03:46 INFO ShutdownHookManager: Shutdown hook called
  35. 23/03/21 14:03:46 INFO ShutdownHookManager: Deleting directory /tmp/spark-1f53cd0a-1ecb-467c-a59c-c84bd2a1196b
  36. 23/03/21 14:03:46 INFO ShutdownHookManager: Deleting directory /tmp/spark-891d26da-45b7-40c5-8e64-a3d2d96b132c

三，分析淘宝数据

  1. 上传数据挂载目录，之后上传到hdfs

  1. ##ubuntu
  2. cd ~
  3. sudo mv taobao_user_behavior_cut.csv /bigdata
  1. cd /usr/local/hadoop
  2. ./bin/hdfs dfs -mkdir taobao
  3. ./bin/hdfs dfs -put /bigdata/taobao_user_behavior_cut.csv taobao

2.数据分析

  • 读取与展示数据
  1. from pyspark.sql import SparkSession
  2. from pyspark.sql.functions import *
  3. from pyspark import SparkContext,StorageLevel
  5. spark = SparkSession.builder.getOrCreate()
  6. raw_data = spark.read.csv("/user/root/taobao/taobao_user_behavior_cut.csv",header=True)
  8. raw_data.show(20)
  1. #将上面的代码写进py文件
  2. cd /bigdata
  3. vim spark_taobao.py
  4. python3 spark_taobao.py

  • 在yarn上处理数据并保存到mysql
  1. from pyspark.sql import SparkSession
  2. from pyspark.sql.functions import *
  3. from pyspark import SparkContext,StorageLevel
  4. import pandas as pd
  6. #读取数据
  7. spark = SparkSession.builder.getOrCreate()
  8. raw_data = spark.read.csv("/user/root/taobao/taobao_user_behavior_cut.csv",header=True)
  10. #删除空缺列，无效列，转化时间，行为，持久化数据
  11. def tran(x):
  12.     tmp = x["time"].split(" ")
  13.     day = tmp[0]
  14.     hour = tmp[1]
  16.     user_id = x["user_id"]
  18.     if int(x["behavior_type"]) ==1 :
  19.         behavior = "点击详情页"
  20.     elif int(x["behavior_type"]) == 2:
  21.         behavior = "收藏"
  22.     elif int(x["behavior_type"]) == 3:
  23.         behavior = "加入购物车"
  24.     else:
  25.         behavior = "购买"
  27.     item_id = x["item_id"]
  28.     item_category = x["item_category"]
  30.     return user_id,item_id,behavior,item_category,day,hour
  32. raw_df = raw_data.drop("user_geohash","_c0").rdd.map(tran).toDF(schema="user_id string,item_id string,behavior string,item_category string,day string,hour string")
  33. raw_df.persist(storageLevel=StorageLevel.MEMORY_AND_DISK)
  35. #计算用户最后一次点击与第一次点击时间的间隔,为了计算留存率，并统计用户购买路径，主要是返回购买前的动作
  36. def countRetention(key,x):
  37.     data = x["day"].values
  38.     id = str(x["user_id"].values[0])
  39.     behavior_values = x["behavior"].values
  40.     behavior_index = len(behavior_values)
  41.     buy_path = "没有购买"
  42.     first_time = [int(data[0].split("-")[1]), int(data[0].split("-")[2])]
  43.     laste_time = [int(data[-1].split("-")[1]), int(data[-1].split("-")[2])]
  45.     if first_time[0] ==11:
  46.         first_time[1] = first_time[1] - 18
  47.     else:
  48.         first_time[1] = 12 + first_time[1]
  50.     if laste_time[0] == 11:
  51.         laste_time[1] = laste_time[1] - 18
  52.     else:
  53.         laste_time[1] = 12 + laste_time[1]
  55.     interval_time = laste_time[1]-first_time[1]
  57.     for index in range(behavior_index):
  58.         if behavior_values[index] == "购买":
  59.             buy_path = behavior_values[index-1]
  61.     return pd.DataFrame([[id, interval_time, buy_path]])
  63. user_retention_data = raw_df.groupBy("user_id").applyInPandas(countRetention,schema="user_id string,retention_date int,buy_path string")
  65. #统计日均流量，月均流量
  66. date_flow = raw_df.groupBy("hour").count()
  67. month_flow = raw_df.groupBy("day").count()
  69. #统计每个商品种类的点击量，购买量
  70. def countClike(key,x):
  71.     behavior = x["behavior"].values
  72.     id = x["item_category"].values[0]
  73.     clike_num = 0
  74.     buy_num = 0
  75.     for i in behavior:
  76.         if i == "购买":
  77.             buy_num += 1
  78.         clike_num += 1
  79.     return pd.DataFrame([[id,clike_num,buy_num]])
  81. click_data = raw_df.select("behavior","item_category").groupBy("item_category").applyInPandas(countClike,schema="item_category string, clike_num int, buy_num int")
  83. #将数据储存进mysql
  84. properties={
  85.     "user" : "root",
  86.     "password" : "mysql123",
  87.     "dirver" : "com.mysql.cj.jdbc.Driver"
  88. }
  90. raw_df.write.jdbc("jdbc:mysql://localhost:3306/taobao_data", table="raw_df", mode="overwrite",properties=properties )
  91. user_retention_data.write.jdbc("jdbc:MySQL://localhost:3306/taobao_data", table="user_retention_data", mode='overwrite',properties=properties )
  92. date_flow.write.jdbc("jdbc:MySQL://localhost:3306/taobao_data", table="date_flow", mode='overwrite',properties=properties )
  93. month_flow.write.jdbc("jdbc:MySQL://localhost:3306/taobao_data", table="month_flow", mode='overwrite',properties=properties )
  94. click_data.write.jdbc("jdbc:MySQL://localhost:3306/taobao_data", table="click_data", mode='overwrite',properties=properties )
  1. #进入mysql，创建数据库taobao_data
  2. mysql -u root -p
  3. create database taobao_data
  4. exit
  6. #打包程序所需依赖包,pyspark等
  7. cp -r /usr/local/lib/python3.10/dist-packages /bigdata
  8. zip spark_taobao.zip dist-packages
  10. #编辑py文件，将上面的python代码写入
  11. vim spark_taobao.py
  13. #上传py文件，在yarn中运行
  14. #要注意executor的内存加起来不能超过实际虚拟机资源，否则yarn会杀掉这个executor
  15. cd /usr/local/spark
  16. ./bin/spark-submit /
  17. --master yarn /
  18. --deploy-mode cluster /
  19. --driver-memory 2g /
  20. --executor-memory 1g /
  21. --executor-cores 1 /
  22. --num-executors 1 /
  23. --py-files='/bigdata/spark_taobao.zip' /
  24. /bigdata/spark_taobao.py
  • 读取mysql数据，分析数据
  1. from pyspark.sql import SparkSession
  2. from pyspark import SparkContext,StorageLevel
  3. from pyspark.sql.functions import *
  4. import pandas
  5. import matplotlib.pyplot as plt
  7. plt.rcParams['font.sans-serif'] = 'simhei'
  8. spark = SparkSession.builder.getOrCreate()
  10. #从mysql上读取数据
  11. url = "jdbc:mysql://localhost:3306/taobao_data"
  12. propertie = {
  13.     "user" : "root",
  14.     "password": "mysql123",
  15.     "dirver" : "com.mysql.cj.jdbc.Driver"
  16. }
  17. user_retention_data = spark.read.jdbc(url=url, table="user_retention_data", properties=propertie)
  18. date_flow = spark.read.jdbc(url=url, table="date_flow", properties=propertie)
  19. month_flow = spark.read.jdbc(url=url, table="month_flow", properties=propertie)
  20. click_data = spark.read.jdbc(url=url, table="click_data", properties=propertie)
  22. #统计7日留存率， 14日留存率， 28日留存率
  23. all_user_num = user_retention_data.count()
  24. seven_retention = user_retention_data.filter(user_retention_data.retention_date >= 7).count()/all_user_num
  25. fourteen_retention = user_retention_data.filter(user_retention_data.retention_date >= 14).count()/all_user_num
  26. te_retention = user_retention_data.filter(user_retention_data.retention_date >= 28).count()/all_user_num
  28. retention_y = [seven_retention, fourteen_retention, te_retention]
  29. retention_x = ["7日留存率", "14日留存率", "28日留存率"]
  31. plt.plot(retention_x, retention_y, color='r', linewidth=2, linestyle='dashdot' )
  33. for x in range(3):
  34.         plt.text(x-0.13,retention_y[x],str(retention_y[x]),ha='center', va= 'bottom',fontsize=9)
  36. plt.savefig("retention.jpg")
  37. plt.clf()
  39. #统计日均流量
  40. date_flow = date_flow.sort("hour").toPandas()
  41. date_flow_x = date_flow["hour"].values
  42. date_flow_y = date_flow["count"].values
  44. plt.figure(figsize=(8,4))
  45. plt.plot(date_flow_x, date_flow_y,color='r', linewidth=2, linestyle='dashdot')
  47. for x in range(24):
  48.         plt.text(x-0.13,date_flow_y[x],str(date_flow_y[x]),ha='center', va= 'bottom',fontsize=9)
  50. plt.savefig("date_flow.jpg")
  51. plt.clf()
  53. #统计月均流量
  54. month_flow = month_flow.sort("day").toPandas()
  55. month_flow_x = month_flow["day"].values
  56. month_flow_y = month_flow["count"].values
  58. plt.figure(figsize=(15,4))
  59. plt.xticks(rotation=90)
  60. plt.plot(month_flow_x, month_flow_y,color='r', linewidth=2, linestyle='dashdot')
  61. plt.savefig("month_flow.jpg",bbox_inches='tight')
  62. plt.clf()
  64. #统计top10的商品
  65. def take(x):
  66.     data_list = []
  67.     for i in range(10):
  68.         data_list.append(x[i])
  69.     return data_list
  71. visit_data = click_data.sort(desc("clike_num")).toPandas()
  72. visit_x = take(visit_data["item_category"].values)
  73. visit_y = take(visit_data["clike_num"].values)
  75. visit_plt = plt.bar(visit_x, visit_y, lw=0.5,fc="b",width=0.5)
  76. plt.bar_label(visit_plt, label_type='edge')
  77. plt.savefig("visit_top10.jpg",bbox_inches='tight')
  78. plt.clf()
  80. buy_data = click_data.sort(desc("buy_num")).toPandas()
  81. buy_x = take(buy_data["item_category"].values)
  82. buy_y = take(buy_data["buy_num"].values)
  84. buy_plt = plt.bar(buy_x, buy_y, lw=0.5,fc="b",width=0.5)
  85. plt.bar_label(buy_plt, label_type='edge')
  86. plt.savefig("buy_top10.jpg",bbox_inches='tight')
  87. plt.clf()
  89. #统计购买路径
  90. buy_path_data = user_retention_data.filter(user_retention_data.buy_path != "没有购买").groupBy("buy_path").count().sort(desc("count")).toPandas()
  91. buy_path_y = buy_path_data["count"].values
  92. buy_path_x= buy_path_data["buy_path"].values
  94. buy_path_plt = plt.bar(buy_path_x, buy_path_y, lw=0.5,fc="b",width=0.5)
  95. plt.bar_label(buy_path_plt, label_type='edge')
  96. plt.savefig("buy_path.jpg",bbox_inches='tight')
  1. #编辑运行py文件，将上面的python代码写入,图片会保存在本目录中
  2. cd /bigdata
  3. vim taobao_analy.py
  4. python3 taobao_analy.py
  • 运行结果
  1. 日均流量，可以看出每晚的21-22点访问人数最多

  1. 月均流量，在12月12日的时候，人数陡然增加，说明促销活动是有效的

  1. 留存率，7日留存率与14日留存率差别不大，但是在28日时陡然下降，可以考虑在这一个周期里做些活动

  1. 访问top10的商品种类

  1. 购买量top10的商品

  1. 用户购买途径，大多数用户都喜欢通过点击详情页进行购买，所以一个好的详情页对于一个商品的销量有着重要影响

四，其他

关于matplotlib中文字体显示问题请参考

findfont: Font family [‘sans-serif‘] not found解决方法_fontfamily[sa_ACE-Mayer的博客-CSDN博客

标签： 大数据 hadoop spark
本文转载自: https://blog.csdn.net/linpaomian/article/details/129687054
版权归原作者 linpaomian 所有， 如有侵权，请联系我们删除。
登录后发布评论

“大数据实战（hadoop+spark+python）：淘宝电商数据分析”的评论:

还没有评论
关于作者
...
overfit同步小助手
文章同步
相关阅读

网络安全法-网络运行安全

使用selenium/drissionpage时如何阻止chrome自动跳转http到https

docker desktop 里部署的Open WebUI 管理员密码忘记了的处理方法

在ubuntu20.04中搭建onsite比赛运行环境

利用开源的低代码表单设计器FcDesigner高效管理和渲染复杂表单结构

Kafka学习笔记

【前端】浏览器输入url到页面呈现发生了什么？