spark = SparkSession.builder.getOrCreate() # 加载Mysql数据 data = spark.read.format("jdbc").options(**options).load() # 将每一行的taglist转为列表 def convert_to_list(line): tmp_list = line[0].replace("#", "").split(",") datas = [] for i in tmp_list: if len(i) > 0 and "牛" not in i: datas.append((i, 1)) return datas rdd = data.rdd.flatMap(lambda line: convert_to_list(line)).reduceByKey(lambda x, y: x + y) schemaString = "tag count" fields = [StructField(field_name, StringType(), True) for field_name in schemaString.split()] schema = StructType(fields) schema_data = spark.createDataFrame(rdd, schema).orderBy("count", ascending=False) # 将数据转换为Pandas数据帧 result_pdf = schema_data.limit(5).toPandas() # 设置matplotlib支持中文 plt.rcParams['font.family'] = ['sans-serif'] plt.rcParams['font.sans-serif'] = ['SimHei'] # colors=color, explode=explode, plt.pie(result_pdf["count"], labels=result_pdf["tag"], shadow=True, autopct='%1.1f%%') plt.legend() plt.show()每行代码讲解
时间: 2024-04-06 12:28:58 浏览: 60
Python库 | dbnd-spark-0.34.7.tar.gz
这段代码主要是用来读取MySQL中的数据,并对数据中的taglist进行处理,将每一行的taglist转换为列表,然后统计列表中每个tag的出现次数,最后将结果排序并输出前五个tag及其对应的出现次数。
具体每行代码的含义如下:
1. `spark = SparkSession.builder.getOrCreate()`:创建一个SparkSession实例,如果已经存在则获取已存在的实例。
2. `data = spark.read.format("jdbc").options(**options).load()`:从MySQL中读取数据,并将数据存储为DataFrame格式。
3. `def convert_to_list(line):`:定义一个转换函数,用来将每行的taglist转换为列表。
4. `tmp_list = line[0].replace("#", "").split(",")`:将每行的taglist去掉#号,并按逗号分割,得到一个tag的列表。
5. `datas = []`:定义一个空列表,用来存储tag及其出现次数。
6. `for i in tmp_list:`:遍历每个tag。
7. `if len(i) > 0 and "牛" not in i:`:如果tag长度大于0且不包含“牛”。
8. `datas.append((i, 1))`:将tag及其出现次数1添加到列表datas中。
9. `return datas`:返回列表datas。
10. `rdd = data.rdd.flatMap(lambda line: convert_to_list(line)).reduceByKey(lambda x, y: x + y)`:将DataFrame转换为RDD,并对RDD中的tag进行统计,得到每个tag及其对应的出现次数。
11. `schemaString = "tag count"`:定义一个字符串,用来表示DataFrame的列名。
12. `fields = [StructField(field_name, StringType(), True) for field_name in schemaString.split()]`:根据列名字符串,定义一个包含两个字段(tag和count)的结构体。
13. `schema = StructType(fields)`:根据结构体定义,创建DataFrame的schema。
14. `schema_data = spark.createDataFrame(rdd, schema).orderBy("count", ascending=False)`:基于RDD和schema,创建DataFrame,并按照count字段进行降序排序。
15. `result_pdf = schema_data.limit(5).toPandas()`:将DataFrame转换为Pandas数据帧,并取前五行数据。
16. `plt.rcParams['font.family'] = ['sans-serif']`:设置matplotlib支持中文。
17. `plt.rcParams['font.sans-serif'] = ['SimHei']`:设置matplotlib使用SimHei字体。
18. `plt.pie(result_pdf["count"], labels=result_pdf["tag"], shadow=True, autopct='%1.1f%%')`:绘制饼图,以tag为标签,以count为数据,并显示百分比。
19. `plt.legend()`:添加图例。
20. `plt.show()`:显示图形。
阅读全文