pySpark数据分析（二）——基于Spark的服装销量分析及可视化
数据库来源：kaggle.com
一、需求分析（对服装销量进行分析）
（一）性别：男女性服装销量；是否成年服装销量
（二）价格：不同价格区间销量
（三）颜色：1、各价格区间衣服颜色销量 2、总体颜色的销量分布
（四）品牌：将销量>x的作为品牌，1、统计各品牌间的销量情况对比；2、品牌与杂牌中的销量情况（牌子数目和总销量对比）
（五）描述：1、商品描述的长短对销量是否有影响 2、商品描述中出现最多的50个字符
（六）图片数量：图片数量对销量的影响
（七）产品名称：统计产品名称长短以及出现最多的20个字符
二、数据存储hive+mysql
使用mysql作为hive的源数据库
https://blog.csdn.net/sh5221180/article/details/106119424
2.1 hadoop配置
1、core-site.xml
2、hdfs-site.xml
格式化hdfs：hdfs namenode -format
通过HADOOP_HOME/sbin/start-dfs.cmd启动hdfs后，可从http://localhost:50070/查看UI界面
error:namenode启动失败：java.lang.UnsatisfiedLinkError: org.apache.hadoop.io.nativeio.NativeIO
解决：需要下载一个hadoop.dll文件
参考文献
2.2 MySQL安装
1、下载：windows版安装地址
2、配置安装：mysql配置
3、启动：service.msc 服务管理启动mysql服务
4、登陆：mysql -uroot -p
5、测试：使用Navicat新建连接进行测试，成功
2.3 配置hive
1、下载：下载地址
2、配置：hive-site.xml配置
3、测试：
启动hadoop：%HADOOP_HOME%\sbin目录下的start-all.cmd，会自动启动hdfs和yarn； 启动hive：%HIVE_HOME%\bin目录下的hive.cmd文件，若缺失需要补充下载。https://www.bjjem.com/article-5545-1.html
2.4 hive连接mysql
1、windows下的hive cmd工具下载，下载schematool.cmd
2、元数据库初始化：hive --service schematool -dbType mysql -initSchema
3、启动hive，测试：show databases
4、从hive数据库下的dbs表可以看到在hdfs://localhost:9000/user/hive/warehouse有hive默认数据库Default Hive database
2.5 hive写入数据
参考链接
1、创建表
create external table if not exists data0 (
ProductID int,
ProductName string,
ProductBrand string,
Gender string,
Price int,
NumImages int,
Description string,
PrimaryColor string
) ROW FORMAT DELIMITED FIELDS TERMINATED BY ',' STORED AS TEXTFILE
2、导入数据
load data local inpath 'F:/Study/spark/dataset/data0.csv' overwrite into table data0;
3、测试查询
select * from data0 limit 3;
2.6 pyspark配置hive
1、将hive-site.xml、hdfs-site.xml、core-site.xml复制进%SPARK_HOME%\conf下
2、下载mysql-connector，将其中的mysql-connector-j-8.0.31.jar包放入%SPARK_HOME%\jar下
3、启动hadoop：%HADOOP_HOME%\sbin\start-all.cmd
4、启动hive：%HIVE_HOME%\bin\hive.cmd
5、驱动器定义，配置hive
# 定义spark驱动器
conf = SparkConf().setMaster("local[2]") \
.setAppName("DataAnalysis").set('spark.executor.memory', '10G')
sc = SparkContext(conf=conf) # 定义驱动器对象，sparksession的子对象
sc.setLogLevel("ERROR") # OFF.FATAL.ERROR.WARN.INFO.DEBUG.TRACE.ALL，设置日志级别
_SPARK_HOST = "spark://spark-master:7077"
_APP_NAME = "test"
spark = SparkSession \
.builder \
.appName("name") \
.enableHiveSupport() \
.getOrCreate() # 启用hive支持
hive_context = HiveContext(spark)
hive_context.sql("use pyspark") # 选择数据库
读取数据
def data_read(hive_table):
"""
读取并返回表数据，打印前五行
"""
hive_read = "select * from {}".format(hive_table)
# 通过SQL语句在hive中查询的数据直接是dataframe的形式
read_df = hive_context.sql(hive_read)
read_df.show(5)
read_df.write.csv(path,mode = "overwrite")
return read_df
error：报错：xxx is not a valid DFS filename
写入数据
def write_data(table_name):
"""
RDD数据转为dataframe，设定列信息并写入hive表
"""
spark.createDataFrame(rdd_disc)\
.toDF("len", "num")
.write.saveAsTable("data_disc",mode = "overwrite")
查看数据
使用nacivat通过SSH连接hive数据库，查看其中的表。
error 1：连接失败：WSAGetLastError return 10060
原因与解决方法：连接hive数据库要启动hive的server2和metastore服务
hive --service metastore &
hive --service hiveserver2 &
dbeaver：https://www.modb.pro/db/327517
error 2：启动server2失败： java.lang.NoSuchMethodError:
原因与解决方法：jar包冲突：http://t.zoukankan.com/guohu-p-13200879.html
三、数据清洗
1、选择子集，去除不要的列
2、列名重命名
3、删除重复值
依据产品ID删除
4、缺失值处理
文本型数据人工填补
5、一致化处理
对数据处理，方便后续分析 异常值处理
四、数据分析
依据需求分析编写代码，将分析后的数据存入hive中。
代码示例：
def data_price(df):
step = 500
max_limit = 10000
rdd_price = df.rdd \
.map(lambda x: x.price) \
.filter(lambda x: x is not None) \
.map(lambda x: max_limit if x > max_limit else x) \
.map(lambda price: (int(price / step), 1)) \
.reduceByKey(lambda x, y: x + y) \
.map(lambda x: (x[0] * step, x[1])) \
.sortByKey(ascending=True)
spark.createDataFrame(rdd_price).toDF("price", "num") \
.write.saveAsTable("data_price", mode="overwrite")
五、可视化
推荐使用微信小程序进行分析结果的展示，由于本人对前端知识不够了解且时间较紧，故使用可视化工具进行结果可视化，具体请参考该工具提供的帮助手册。
1、启动hadoop
2、启动hive
3、hive启动server2服务，使得数据库支持jdbc连接
hive.cmd --service hiveserver2 start
hive --service metastore &
4、建立连接（hive连接失败）
error：required-field-serverprotocolversion-is-unset
最后做法是将数据存入本地mysql数据库，使用可视化工具进行连接并使用适当形式进行展示。
参考文章
发表评论