【Python-Spark（大规模数据）】

Python-Spark（大规模数据）

  • ■ Spark
  • ■ PySparl编程模型
  • ■ 基础准备
  • ■ 数据输入
  • ■ RDD的map成员方法的使用
  • ■ RDD的flatMap成员方法的使用
  • ■ RDD的reduceByKey成员方法的使用
  • ■ 单词计数统计
  • ■ RDD的filter成员方法的使用
  • ■ RDD的distinct成员方法的使用
  • ■ RDD的sortBy成员方法的使用
  • ■ 案例：JSON商品统计
  • ■ 将RDD输出为Python对象
  • ■ 将RDD输出到文件中
  • ■ PySpark综合案例
  • ■ PySpark综合案例

■ Spark

Apache Spark 是用于大规模数据处理的统一分析引擎。
PySpark是由Spark官方开发的Python语言第三方库。

■ PySparl编程模型

  • 通过SparkContext对象,完成数据输入
  • 输入数据后得到RDD对象,对RDD对象进行迭代计算
  • 最终通过RDD对象的成员方法,完成数据输出工作
    在这里插入图片描述

■ 基础准备

# 导包
from pyspark import SparkConf, SparkContext
# 创建SparkConf类对象
conf = SparkConf().setMaster("local[*]").setAppName("test_spark_app")
# 基于SparkConf类对象创建SparkContext对象
sc = SparkContext(conf=conf)
# 打印PySpark的运行版本
print(sc.version)
# 停止SparkContext对象的运行（停止PySpark程序）
sc.stop()

■ 数据输入

"""
演示通过PySpark代码加载数据，即数据输入
"""
from pyspark import SparkConf, SparkContextconf = SparkConf().setMaster("local[*]").setAppName("test_spark")
sc = SparkContext(conf=conf)# # 通过parallelize方法将Python对象加载到Spark内，成为RDD对象
# rdd1 = sc.parallelize([1, 2, 3, 4, 5])
# rdd2 = sc.parallelize((1, 2, 3, 4, 5))
# rdd3 = sc.parallelize("abcdefg")
# rdd4 = sc.parallelize({1, 2, 3, 4, 5})
# rdd5 = sc.parallelize({"key1": "value1", "key2": "value2"})
#
# # 如果要查看RDD里面有什么内容，需要用collect()方法
# print(rdd1.collect())
# print(rdd2.collect())
# print(rdd3.collect())
# print(rdd4.collect())
# print(rdd5.collect())# 用过textFile方法，读取文件数据加载到Spark内，成为RDD对象
rdd = sc.textFile("D:/hello.txt")
print(rdd.collect())
rdd.map()
sc.stop()

■ RDD的map成员方法的使用

"""
演示RDD的map成员方法的使用
"""
from pyspark import SparkConf, SparkContext
import os
os.environ['PYSPARK_PYTHON'] = "D:/dev/python/python310/python.exe"
conf = SparkConf().setMaster("local[*]").setAppName("test_spark")
sc = SparkContext(conf=conf)# 准备一个RDD
rdd = sc.parallelize([1, 2, 3, 4, 5])
# 通过map方法将全部数据都乘以10
# def func(data):
#     return data * 10rdd2 = rdd.map(lambda x: x * 10).map(lambda x: x + 5)print(rdd2.collect())
# (T) -> U
# (T) -> T# 链式调用

■ RDD的flatMap成员方法的使用

"""
演示RDD的flatMap成员方法的使用
"""
from pyspark import SparkConf, SparkContext
import os
os.environ['PYSPARK_PYTHON'] = "D:/dev/python/python310/python.exe"
conf = SparkConf().setMaster("local[*]").setAppName("test_spark")
sc = SparkContext(conf=conf)# 准备一个RDD
rdd = sc.parallelize(["itheima itcast 666", "itheima itheima itcast", "python itheima"])# 需求，将RDD数据里面的一个个单词提取出来
rdd2 = rdd.flatMap(lambda x: x.split(" "))
print(rdd2.collect())

■ RDD的reduceByKey成员方法的使用

"""
演示RDD的reduceByKey成员方法的使用
"""
from pyspark import SparkConf, SparkContext
import os
os.environ['PYSPARK_PYTHON'] = "D:/dev/python/python310/python.exe"
conf = SparkConf().setMaster("local[*]").setAppName("test_spark")
sc = SparkContext(conf=conf)# 准备一个RDD
rdd = sc.parallelize([('男', 99), ('男', 88), ('女', 99), ('女', 66)])
# 求男生和女生两个组的成绩之和
rdd2 = rdd.reduceByKey(lambda a, b: a + b)
print(rdd2.collect())

■ 单词计数统计

"""
完成练习案例：单词计数统计
"""# 1. 构建执行环境入口对象
from pyspark import SparkContext, SparkConf
import os
os.environ['PYSPARK_PYTHON'] = "D:/dev/python/python310/python.exe"
conf = SparkConf().setMaster("local[*]").setAppName("test_spark")
sc = SparkContext(conf=conf)
# 2. 读取数据文件
rdd = sc.textFile("D:/hello.txt")
# 3. 取出全部单词
word_rdd = rdd.flatMap(lambda x: x.split(" "))
# 4. 将所有单词都转换成二元元组，单词为Key，value设置为1
word_with_one_rdd = word_rdd.map(lambda word: (word, 1))
# 5. 分组并求和
result_rdd = word_with_one_rdd.reduceByKey(lambda a, b: a + b)
# 6. 打印输出结果
print(result_rdd.collect())

■ RDD的filter成员方法的使用

"""
演示RDD的filter成员方法的使用
"""
from pyspark import SparkConf, SparkContext
import os
os.environ['PYSPARK_PYTHON'] = "D:/dev/python/python310/python.exe"
conf = SparkConf().setMaster("local[*]").setAppName("test_spark")
sc = SparkContext(conf=conf)# 准备一个RDD
rdd = sc.parallelize([1, 2, 3, 4, 5])
# 对RDD的数据进行过滤
rdd2 = rdd.filter(lambda num: num % 2 == 0)print(rdd2.collect())

■ RDD的distinct成员方法的使用

"""
演示RDD的distinct成员方法的使用
"""
from pyspark import SparkConf, SparkContext
import os
os.environ['PYSPARK_PYTHON'] = "D:/dev/python/python310/python.exe"
conf = SparkConf().setMaster("local[*]").setAppName("test_spark")
sc = SparkContext(conf=conf)# 准备一个RDD
rdd = sc.parallelize([1, 1, 3, 3, 5, 5, 7, 8, 8, 9, 10])
# 对RDD的数据进行去重
rdd2 = rdd.distinct()print(rdd2.collect())

■ RDD的sortBy成员方法的使用

"""
演示RDD的sortBy成员方法的使用
"""
from pyspark import SparkConf, SparkContext
import os
os.environ['PYSPARK_PYTHON'] = "D:/dev/python/python310/python.exe"
conf = SparkConf().setMaster("local[*]").setAppName("test_spark")
sc = SparkContext(conf=conf)# 1. 读取数据文件
rdd = sc.textFile("D:/hello.txt")
# 2. 取出全部单词
word_rdd = rdd.flatMap(lambda x: x.split(" "))
# 3. 将所有单词都转换成二元元组，单词为Key，value设置为1
word_with_one_rdd = word_rdd.map(lambda word: (word, 1))
# 4. 分组并求和
result_rdd = word_with_one_rdd.reduceByKey(lambda a, b: a + b)
# 5. 对结果进行排序
final_rdd = result_rdd.sortBy(lambda x: x[1], ascending=True, numPartitions=1)
print(final_rdd.collect())

■ 案例：JSON商品统计

"""
完成练习案例：JSON商品统计
需求：
1. 各个城市销售额排名，从大到小
2. 全部城市，有哪些商品类别在售卖
3. 北京市有哪些商品类别在售卖
"""
from pyspark import SparkConf, SparkContext
import os
import json
os.environ['PYSPARK_PYTHON'] = 'D:/dev/python/python310/python.exe'
conf = SparkConf().setMaster("local[*]").setAppName("test_spark")
sc = SparkContext(conf=conf)# TODO 需求1： 城市销售额排名
# 1.1 读取文件得到RDD
file_rdd = sc.textFile("D:/orders.txt")
# 1.2 取出一个个JSON字符串
json_str_rdd = file_rdd.flatMap(lambda x: x.split("|"))
# 1.3 将一个个JSON字符串转换为字典
dict_rdd = json_str_rdd.map(lambda x: json.loads(x))
# 1.4 取出城市和销售额数据
# (城市，销售额)
city_with_money_rdd = dict_rdd.map(lambda x: (x['areaName'], int(x['money'])))
# 1.5 按城市分组按销售额聚合
city_result_rdd = city_with_money_rdd.reduceByKey(lambda a, b: a + b)
# 1.6 按销售额聚合结果进行排序
result1_rdd = city_result_rdd.sortBy(lambda x: x[1], ascending=False, numPartitions=1)
print("需求1的结果：", result1_rdd.collect())
# TODO 需求2： 全部城市有哪些商品类别在售卖
# 2.1 取出全部的商品类别
category_rdd = dict_rdd.map(lambda x: x['category']).distinct()
print("需求2的结果：", category_rdd.collect())
# 2.2 对全部商品类别进行去重
# TODO 需求3： 北京市有哪些商品类别在售卖
# 3.1 过滤北京市的数据
beijing_data_rdd = dict_rdd.filter(lambda x: x['areaName'] == '北京')
# 3.2 取出全部商品类别
result3_rdd = beijing_data_rdd.map(lambda x: x['category']).distinct()
print("需求3的结果：", result3_rdd.collect())
# 3.3 进行商品类别去重

■ 将RDD输出为Python对象

"""
演示将RDD输出为Python对象
"""from pyspark import SparkConf, SparkContext
import os
import json
os.environ['PYSPARK_PYTHON'] = 'D:/dev/python/python310/python.exe'
conf = SparkConf().setMaster("local[*]").setAppName("test_spark")
sc = SparkContext(conf=conf)# 准备RDD
rdd = sc.parallelize([1, 2, 3, 4, 5])# collect算子，输出RDD为list对象
rdd_list: list = rdd.collect()
print(rdd_list)
print(type(rdd_list))
# reduce算子，对RDD进行两两聚合
num = rdd.reduce(lambda a, b: a + b)
print(num)
# take算子，取出RDD前N个元素，组成list返回
take_list = rdd.take(3)
print(take_list)
# count，统计rdd内有多少条数据，返回值为数字
num_count = rdd.count()
print(f"rdd内有{num_count}个元素")sc.stop()

■ 将RDD输出到文件中

"""
演示将RDD输出到文件中
"""from pyspark import SparkConf, SparkContext
import os
import json
os.environ['PYSPARK_PYTHON'] = 'D:/dev/python/python310/python.exe'
os.environ['HADOOP_HOME'] = "D:/dev/hadoop-3.0.0"
conf = SparkConf().setMaster("local[*]").setAppName("test_spark")sc = SparkContext(conf=conf)# 准备RDD1
rdd1 = sc.parallelize([1, 2, 3, 4, 5], numSlices=1)# 准备RDD2
rdd2 = sc.parallelize([("Hello", 3), ("Spark", 5), ("Hi", 7)], 1)# 准备RDD3
rdd3 = sc.parallelize([[1, 3, 5], [6, 7, 9], [11, 13, 11]], 1)# 输出到文件中
rdd1.saveAsTextFile("D:/output1")
rdd2.saveAsTextFile("D:/output2")
rdd3.saveAsTextFile("D:/output3")

■ PySpark综合案例

"""
演示PySpark综合案例
"""from pyspark import SparkConf, SparkContext
import os
import json
os.environ['PYSPARK_PYTHON'] = 'D:/dev/python/python310/python.exe'
os.environ['HADOOP_HOME'] = "D:/dev/hadoop-3.0.0"
conf = SparkConf().setMaster("local[*]").setAppName("test_spark")
conf.set("spark.default.parallelism", "1")
sc = SparkContext(conf=conf)# 读取文件转换成RDD
file_rdd = sc.textFile("D:/search_log.txt")
# TODO 需求1： 热门搜索时间段Top3（小时精度）
# 1.1 取出全部的时间并转换为小时
# 1.2 转换为(小时, 1) 的二元元组
# 1.3 Key分组聚合Value
# 1.4 排序（降序）
# 1.5 取前3
result1 = file_rdd.map(lambda x: (x.split("\t")[0][:2], 1)).\reduceByKey(lambda a, b: a + b).\sortBy(lambda x: x[1], ascending=False, numPartitions=1).\take(3)
print("需求1的结果：", result1)# TODO 需求2： 热门搜索词Top3
# 2.1 取出全部的搜索词
# 2.2 (词, 1) 二元元组
# 2.3 分组聚合
# 2.4 排序
# 2.5 Top3
result2 = file_rdd.map(lambda x: (x.split("\t")[2], 1)).\reduceByKey(lambda a, b: a + b).\sortBy(lambda x: x[1], ascending=False, numPartitions=1).\take(3)
print("需求2的结果：", result2)# TODO 需求3： 统计黑马程序员关键字在什么时段被搜索的最多
# 3.1 过滤内容，只保留黑马程序员关键词
# 3.2 转换为(小时, 1) 的二元元组
# 3.3 Key分组聚合Value
# 3.4 排序（降序）
# 3.5 取前1
result3 = file_rdd.map(lambda x: x.split("\t")).\filter(lambda x: x[2] == '黑马程序员').\map(lambda x: (x[0][:2], 1)).\reduceByKey(lambda a, b: a + b).\sortBy(lambda x: x[1], ascending=False, numPartitions=1).\take(1)
print("需求3的结果：", result3)# TODO 需求4： 将数据转换为JSON格式，写出到文件中
# 4.1 转换为JSON格式的RDD
# 4.2 写出为文件
file_rdd.map(lambda x: x.split("\t")).\map(lambda x: {"time": x[0], "user_id": x[1], "key_word": x[2], "rank1": x[3], "rank2": x[4], "url": x[5]}).\saveAsTextFile("D:/output_json")

■ PySpark综合案例

"""
演示PySpark综合案例
"""from pyspark import SparkConf, SparkContext
import os
os.environ['PYSPARK_PYTHON'] = '/export/server/anaconda3/bin/python'
os.environ['HADOOP_HOME'] = "/export/server/hadoop-3.3.1"
conf = SparkConf().setAppName("spark_cluster")
conf.set("spark.default.parallelism", "24")
sc = SparkContext(conf=conf)# 读取文件转换成RDD
file_rdd = sc.textFile("hdfs://m1:8020/data/search_log.txt")
# TODO 需求1： 热门搜索时间段Top3（小时精度）
# 1.1 取出全部的时间并转换为小时
# 1.2 转换为(小时, 1) 的二元元组
# 1.3 Key分组聚合Value
# 1.4 排序（降序）
# 1.5 取前3
result1 = file_rdd.map(lambda x: (x.split("\t")[0][:2], 1)).\reduceByKey(lambda a, b: a + b).\sortBy(lambda x: x[1], ascending=False, numPartitions=1).\take(3)
print("需求1的结果：", result1)# TODO 需求2： 热门搜索词Top3
# 2.1 取出全部的搜索词
# 2.2 (词, 1) 二元元组
# 2.3 分组聚合
# 2.4 排序
# 2.5 Top3
result2 = file_rdd.map(lambda x: (x.split("\t")[2], 1)).\reduceByKey(lambda a, b: a + b).\sortBy(lambda x: x[1], ascending=False, numPartitions=1).\take(3)
print("需求2的结果：", result2)# TODO 需求3： 统计黑马程序员关键字在什么时段被搜索的最多
# 3.1 过滤内容，只保留黑马程序员关键词
# 3.2 转换为(小时, 1) 的二元元组
# 3.3 Key分组聚合Value
# 3.4 排序（降序）
# 3.5 取前1
result3 = file_rdd.map(lambda x: x.split("\t")).\filter(lambda x: x[2] == '黑马程序员').\map(lambda x: (x[0][:2], 1)).\reduceByKey(lambda a, b: a + b).\sortBy(lambda x: x[1], ascending=False, numPartitions=1).\take(1)
print("需求3的结果：", result3)# TODO 需求4： 将数据转换为JSON格式，写出到文件中
# 4.1 转换为JSON格式的RDD
# 4.2 写出为文件
file_rdd.map(lambda x: x.split("\t")).\map(lambda x: {"time": x[0], "user_id": x[1], "key_word": x[2], "rank1": x[3], "rank2": x[4], "url": x[5]}).\saveAsTextFile("hdfs://m1:8020/output/output_json")

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.hqwc.cn/news/641468.html

如若内容造成侵权/违法违规/事实不符，请联系编程知识网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

【每日力扣】41. 缺失的第一个正数 238. 除自身以外数组的乘积 189. 轮转数组

【每日力扣】41. 缺失的第一个正数 238. 除自身以外数组的乘积 189. 轮转数组

🔥 个人主页: 黑洞晓威 😀你不必等到非常厉害，才敢开始，你需要开始，才会变的非常厉害 41. 缺失的第一个正数 给你一个未排序的整数数组 nums ，请你找出其中没有出现的最小的正整数。 请你实现时间复杂度为…
阅读更多...
0.5W 3KVDC 隔离单、双输出 DC/DC 电源模块——TPV-W5 24V 48V 系列

0.5W 3KVDC 隔离单、双输出 DC/DC 电源模块——TPV-W5 24V 48V 系列

TPV-W5系列提供正负双输出和单输出，工业级环境温度，用于PCB安装的国际标准结构。此系列产品小巧，效率高，低输出纹波及能承受3000V以上的耐压，用于需要正负电压或单输出和高隔离电压的场合。封装有SIP和DIP可选。
阅读更多...
快刀斩乱麻，DevOps让代码评审也自动起来

快刀斩乱麻，DevOps让代码评审也自动起来

​在 Dr.Michaela Greiler 的 How Code Reviews at Microsoft 一文中提到，微软有 140000 名员工，其中 44%员工是工程师。这意味着，有超过 6000 名的工程师同时在同一个代码库上开发 Office、Visual Studio、Windows 等产品。 想要确保不同子…
阅读更多...
面试宝典（1）——数据库篇（MySQL）

面试宝典（1）——数据库篇（MySQL）

面试宝典（1）——数据库篇（MySQL） 1.什么是索引？ 索引是一种用于加快数据库查询速度的数据结构。 索引可以帮助数据库快速定位到数据库表中特定列的记录，从而加快数据检索和查询的速度。 通过在表的列上…
阅读更多...
使用Azure AI Search和LlamaIndex构建高级RAG应用

使用Azure AI Search和LlamaIndex构建高级RAG应用

RAG 是一种将公司信息合并到基于大型语言模型 （LLM） 的应用程序中的常用方法。借助 RAG，AI 应用程序可以近乎实时地访问最新信息，团队可以保持对其数据的控制。 在 RAG 中，您可以评估和修改各个阶段以改进结果&#x…
阅读更多...
Web3钱包开发获取测试币-Base Sepolia(二)

Web3钱包开发获取测试币-Base Sepolia(二)

Web3钱包开发获取测试币-Base Sepolia(二) ![在这里插入图片描述](https://img-blog.csdnimg.cn/direct/b0c0ac86b04a496087471388532bc54a.png) 基于上篇 Web3钱包开发获取测试币-Polygon Mumbai(一) ：https://suwu150.blog.csdn.net/article/details/137949473 我…
阅读更多...
Git 仓库内容操作

Git 仓库内容操作

Git 仓库内容操作 | CoderMast编程桅杆Git 仓库内容操作 添加文件到暂存区 使用如下指令将工作区的文件添加到暂存区，告诉 Git 在下次 commit 时哪些文件做出了修改。 commit 指令详看后续 添加一个或多个文件到暂存区： 添加指定目录到暂存区 添加当前目…
阅读更多...
使用 ArcGIS 对洪水预测进行建模

使用 ArcGIS 对洪水预测进行建模

第一步 — 下载数据 所有数据均已包含在 Esri 提供的项目压缩文件中。我将创建一个名为“Stowe_Hydrology.gdb”的新地理数据库，在其中保存这些数据以及创建的所有后续图层。 1-0。斯托市边界 斯托城市边界是佛蒙特州地理信息中心提供的矢量要素类面。我将这一层称为“Stow…
阅读更多...
2018年华三杯山东省赛决赛实验

2018年华三杯山东省赛决赛实验

2018年华三杯山东省赛决赛实验 拓扑图 配置需求 请考生根据以下配置需求在 HCL中的设备上进行相关配置。 网络设备虚拟化 数据中心交换机需要实现虚拟化。支持的虚拟化技术 IRF,所配置的参数要求如下: 链形堆叠,IRF Domain 值为 10; IRF1的 member ID 为 1,IRF2的 member …
阅读更多...
【MySQL】Linux环境下MySQL基本操作

【MySQL】Linux环境下MySQL基本操作

目录 一、登录进入MySQL 二、MySQL数据库的查看、创建、删除、重命名、拷贝操作 三、数据库下表的创建、删除、查看、修改（增加、删除、更新字段/列，修改字段/列名） 四、表中数据的插入、删除、查找、更新 一、登录进入MySQL mysql -u u…
阅读更多...
Spring的过滤器、拦截器、切面区别及案例分析

Spring的过滤器、拦截器、切面区别及案例分析

Spring的过滤器、拦截器、切面 三者的区别，以及对应案例分析 一、三者的实现方式 1.1 过滤器 xxxFilter 过滤器的配置比较简单，直接实现Filter接口即可，也可以通过WebFilter注解实现对特定URL的拦截，Filter接口中定义了三个方法…
阅读更多...
Java 网络编程之TCP(一)：基于BIO

Java 网络编程之TCP(一)：基于BIO

环境： jdk 17 IntelliJ IDEA 2023.1.1 (Ultimate Edition) Windows 10 专业版 22H2 TCP：面向连接的，可靠的数据传送协议 Java中的TCP网络编程，其实就是基于常用的BIO和NIO来实现的，本文先讨论BIO； BIO…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023