当前位置:   article > 正文

【PySpark】Python 中进行大规模数据处理和分析_python spark数据分析

python spark数据分析

一、前言介绍
二、基础准备
三、数据输入
四、数据计算
五、数据输出
六、分布式集群运行

一、前言介绍

Spark概述

Apache Spark 是一个开源的大数据处理框架,提供了高效、通用、分布式的大规模数据处理能力。Spark 的主要特点包括:

  1. 速度快:
    Spark 提供了内存计算功能,相较于传统的批处理框架(如Hadoop MapReduce),Spark 能够更高效地执行数据处理任务。Spark 将中间数据存储在内存中,减少了磁盘 I/O,从而加速了计算过程。

  2. 通用性:
    Spark 提供了用于批处理、交互式查询、流处理和机器学习等多种计算模式的 API。这种通用性使得 Spark 在不同的数据处理场景中都能发挥作用。

  3. 易用性:
    Spark 提供了易于使用的高级 API,其中最为知名的是 Spark SQL 和 DataFrame API。这些 API 可以让用户用 SQL 查询语言或类似于 Pandas 的操作方式对数据进行处理,降低了使用门槛。

  4. 弹性计算:
    Spark 可以在集群中分布式执行计算任务,充分利用集群中的计算资源。它具有自动容错和任务重启的机制,保障了计算的稳定性。

  5. 丰富的生态系统:
    Spark 生态系统包括 Spark SQL、Spark Streaming、MLlib(机器学习库)、GraphX(图计算库)等模块,提供了全面的大数据处理解决方案。

Spark 的核心概念包括:

  • RDD(Resilient Distributed Dataset): RDD 是 Spark 中的基本数据抽象,代表分布式的不可变的数据集。Spark 的所有计算都是基于 RDD 进行的。

  • DataFrame: DataFrame 是 Spark 2.0 引入的一种抽象数据结构,提供了类似于关系型数据库表的操作接口。DataFrame 可以通过 Spark SQL 进行查询和操作。

  • Spark SQL: Spark SQL 提供了用于在 Spark 上进行结构化数据处理的 API。它支持 SQL 查询、DataFrame 操作和集成 Hive 查询等。

  • Spark Streaming: Spark Streaming 允许以流式的方式处理实时数据,提供了类似于批处理的 API。

  • MLlib: MLlib 是 Spark 的机器学习库,提供了一系列常见的机器学习算法和工具,方便用户进行大规模机器学习任务。

  • GraphX: GraphX 是 Spark 的图计算库,用于处理大规模图数据。

总体而言,Spark 是一个灵活、强大且易于使用的大数据处理框架,适用于各种规模的数据处理和分析任务。

PySpark概述

PySpark 是 Apache Spark 的 Python API,用于在 Python 中进行大规模数据处理和分析。Spark 是一个用于快速、通用、分布式计算的开源集群计算系统,而 PySpark 则是 Spark 的 Python 版本。

以下是使用 PySpark 进行基本操作的简要步骤:

  1. 安装 PySpark:
    使用以下命令安装 PySpark:

    pip install pyspark
    
    • 1
  2. 创建 SparkSession:
    在 PySpark 中,SparkSession 是与 Spark 进行交互的入口。可以使用以下代码创建一个 SparkSession

    from pyspark.sql import SparkSession
    
    # 创建 SparkSession
    spark = SparkSession.builder.appName("example").getOrCreate()
    
    • 1
    • 2
    • 3
    • 4
  3. 读取数据:
    PySpark 提供了用于读取不同数据源的 API。以下是从文本文件读取数据的示例:

    # 从文本文件读取数据
    data = spark.read.text("path/to/textfile")
    
    • 1
    • 2
  4. 数据转换和处理:
    使用 PySpark 的 DataFrame API 进行数据转换和处理。DataFrame 是一个类似于表的数据结构,可以进行 SQL 风格的查询和操作。

    # 展示 DataFrame 的前几行数据
    data.show()
    
    # 进行数据筛选
    filtered_data = data.filter(data["column"] > 10)
    
    • 1
    • 2
    • 3
    • 4
    • 5
  5. 执行 SQL 查询:
    使用 PySpark 提供的 SQL 接口,可以在 DataFrame 上执行 SQL 查询。

    # 创建临时视图
    data.createOrReplaceTempView("my_table")
    
    # 执行 SQL 查询
    result = spark.sql("SELECT * FROM my_table WHERE column > 10")
    
    • 1
    • 2
    • 3
    • 4
    • 5
  6. 保存结果:
    将处理后的结果保存到文件或其他数据源。

    # 保存到文本文件
    result.write.text("path/to/output")
    
    • 1
    • 2
  7. 关闭 SparkSession:
    在完成所有操作后,关闭 SparkSession。

    # 关闭 SparkSession
    spark.stop()
    
    • 1
    • 2

以上是一个简单的 PySpark 示例。实际应用中,可以根据具体需求使用更多功能,例如连接不同数据源、使用机器学习库(MLlib)进行机器学习任务等。 PySpark 提供了强大的工具和库,适用于大规模数据处理和分析的场景。

Spark作为全球顶级的分布式计算框架,支持众多的编程语言进行开发。
而Python语言,则是Spark重点支持的方向。

二、基础准备

1、PySpark库的安装

同其它的Python第三方库一样,PySpark同样可以使用pip程序进行安装。

在”CMD”命令提示符程序内,输入:

pip install pyspark
  • 1

或使用国内代理镜像网站(清华大学源)

pip install -i https://pypi.tuna.tsinghua.edu.cn/simple pyspark
  • 1

2、PySpark执行环境入口对象的构建

想要使用PySpark库完成数据处理,首先需要构建一个执行环境入口对象。
PySpark的执行环境入口对象是:类 SparkContext 的类对象

"""
演示获取PySpark的执行环境入库对象:SparkContext
并通过SparkContext对象获取当前PySpark的版本
"""

# 导包
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()
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15

3、PySpark的编程模型

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

总结

  1. 如何安装PySpark库
    pip install pyspark
  2. 为什么要构建SparkContext对象作为执行入口
    PySpark的功能都是从SparkContext对象作为开始
  3. PySpark的编程模型是?
    • 数据输入:通过SparkContext完成数据读取
    • 数据计算:读取到的数据转换为RDD对象,调用RDD的成员方法完成计算
    • 数据输出:调用RDD的数据输出相关成员方法,将结果输出到list、元组、字典、文本文件、数据库等

三、数据输入

RDD对象

如图可见,PySpark支持多种数据的输入,在输入完成后,都会得到一个:RDD类的对象
RDD全称为:弹性分布式数据集(Resilient Distributed Datasets)
PySpark针对数据的处理,都是以RDD对象作为载体,即:

  • 数据存储在RDD内
  • 各类数据的计算方法,也都是RDD的成员方法
  • RDD的数据计算方法,返回值依旧是RDD对象
    在这里插入图片描述
    PySpark的编程模型(上图)可以归纳为:
  • 准备数据到RDD -> RDD迭代计算 -> RDD导出为list、文本文件等
  • 即:源数据 -> RDD -> 结果数据

PySpark数据输入的2种方法

Python数据容器转RDD对象在这里插入图片描述

读取文件转RDD对象

在这里插入图片描述

总结

  1. RDD对象是什么?为什么要使用它?
  • RDD对象称之为分布式弹性数据集,是PySpark中数据计算的载体,它可以:

    • 提供数据存储
    • 提供数据计算的各类方法
    • 数据计算的方法,返回值依旧是RDD(RDD迭代计算)
  • 后续对数据进行各类计算,都是基于RDD对象进行

  1. 如何输入数据到Spark(即得到RDD对象)
    • 通过SparkContext的parallelize成员方法,将Python数据容器转换为RDD对象
    • 通过SparkContext的textFile成员方法,读取文本文件得到RDD对象

四、数据计算

1、map方法

PySpark的数据计算,都是基于RDD对象来进行的,那么如何进行呢?
自然是依赖,RDD对象内置丰富的:成员方法(算子)
在这里插入图片描述

"""
演示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 * 10

rdd2 = rdd.map(lambda x: x * 10).map(lambda x: x + 5)

print(rdd2.collect())
# (T) -> U
# (T) -> T

# 链式调用
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22

在这里插入图片描述

总结

  1. map算子(成员方法)
    接受一个处理函数,可用lambda表达式快速编写
    对RDD内的元素逐个处理,并返回一个新的RDD
  2. 链式调用
    对于返回值是新RDD的算子,可以通过链式调用的方式多次调用算子。

2、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())
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15

总结

flatMap算子

  • 计算逻辑和map一样
  • 可以比map多出,解除一层嵌套的功能

3、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
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15

总结

reduceByKey算子
接受一个处理函数,对数据进行两两计算
在这里插入图片描述

WordCount案例

"""
完成练习案例:单词计数统计
"""

# 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())
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20

4、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())
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15

总结

filter算子

  • 接受一个处理函数,可用lambda快速编写
  • 函数对RDD数据逐个处理,得到True的保留至返回值的RDD中

5、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())
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15

总结

distinct算子
完成对RDD内数据的去重操作

6、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())
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20

总结

sortBy算子

  • 接收一个处理函数,可用lambda快速编写
  • 函数表示用来决定排序的依据
  • 可以控制升序或降序
  • 全局排序需要设置分区数为1

五、数据输出

1、输出为Python对象

将RDD的结果输出为Python对象的各类方法
在这里插入图片描述

collect方法

在这里插入图片描述

reduce方法

在这里插入图片描述

take方法

在这里插入图片描述

count方法

在这里插入图片描述

总结

  1. Spark的编程流程就是:
    • 将数据加载为RDD(数据输入)
    • 对RDD进行计算(数据计算)
    • 将RDD转换为Python对象(数据输出)
  2. 数据输出的方法
    • collect:将RDD内容转换为list
    • reduce:对RDD内容进行自定义聚合
    • take:取出RDD的前N个元素组成list
    • count:统计RDD元素个数
      数据输出可用的方法是很多的。

2、输出到文件中

将RDD的内容输出到文件中

saveAsTextFile方法

在这里插入图片描述

注意事项

调用保存文件的算子,需要配置Hadoop依赖

更改RDD的分区数为1

在这里插入图片描述

总结

  1. RDD输出到文件的方法
    • rdd.saveAsTextFile(路径)
    • 输出的结果是一个文件夹
    • 有几个分区就输出多少个结果文件
  2. 如何修改RDD分区
    • SparkConf对象设置conf.set(“spark.default.parallelism”, “1”)
    • 创建RDD的时候,sc.parallelize方法传入numSlices参数为1

六、分布式集群运行

在 Spark 中,分布式集群运行是其强大性能的体现。下面是使用 Spark 进行分布式集群运行的基本步骤:

  1. 准备 Spark 安装:
    在集群中的每台机器上安装 Spark。确保每台机器都能访问相同的 Spark 安装路径。

  2. 配置 Spark:
    在 Spark 安装路径下,编辑 conf/spark-env.sh 文件,设置一些必要的环境变量,例如 Java 路径、Spark 主节点地址等。确保所有节点的配置文件保持一致。

  3. 启动 Spark 主节点(Master):
    在集群中选择一台机器作为 Spark 主节点,执行以下命令启动主节点:

    sbin/start-master.sh
    
    • 1

    默认情况下,主节点的 Web UI 地址是 http://localhost:8080

  4. 启动 Spark 工作节点(Worker):
    在其余机器上执行以下命令启动工作节点,将它们连接到主节点:

    sbin/start-worker.sh spark://<master-node-ip>:<port>
    
    • 1

    <master-node-ip> 是主节点的 IP 地址,<port> 是主节点的端口号(默认为 7077)。

  5. 提交 Spark 应用程序:
    编写 Spark 应用程序,并使用以下命令提交到 Spark 集群:

    bin/spark-submit --class com.example.MyApp --master spark://<master-node-ip>:<port> myapp.jar
    
    • 1

    com.example.MyApp 是你的应用程序主类,myapp.jar 是打包好的应用程序 JAR 文件。

  6. 监控和调优:
    可以通过 Spark 的 Web UI(默认地址为 http://localhost:4040)监控集群运行状态,查看任务的执行情况、资源使用情况等。根据实际情况进行性能调优。

  7. 停止 Spark 集群:
    当任务执行完成后,可以停止 Spark 集群。首先停止工作节点:

    sbin/stop-worker.sh
    
    • 1

    然后停止主节点:

    sbin/stop-master.sh
    
    • 1

这些步骤涵盖了在分布式集群上运行 Spark 应用程序的基本流程。确保配置正确、节点正常连接,以及应用程序能够充分利用集群中的计算资源。 Spark 提供了灵活的配置选项,可以根据具体的集群规模和需求进行调整。

将案例提交到YARN集群中运行

提交命令:

bin/spark-submit --master yarn --num-executors 3 --queue root.teach --executor-cores 4 --executor-memory 4g /home/hadoop/demo.py
  • 1

上面的 Spark 提交命令已经包括了提交到 YARN 集群的必要参数。
以下是命令的解释:

bin/spark-submit
--master yarn                # 指定 Spark 的主节点为 YARN
--num-executors 3             # 指定执行器的数量
--queue root.teach            # 指定 YARN 队列
--executor-cores 4            # 指定每个执行器的核心数
--executor-memory 4g          # 指定每个执行器的内存大小
/home/hadoop/demo.py          # 提交的 Spark 应用程序的路径
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7

解释一下每个参数的作用:

  • --master yarn: 指定 Spark 的主节点为 YARN。这告诉 Spark 将任务提交到 YARN 集群管理器。

  • --num-executors 3: 指定执行器的数量。这是 YARN 上的计算资源,即分配给 Spark 应用程序的节点数量。

  • --queue root.teach: 指定 YARN 队列。这是一个可选的参数,用于将 Spark 应用程序提交到指定的 YARN 队列。

  • --executor-cores 4: 指定每个执行器的核心数。这告诉 YARN 每个执行器可以使用的 CPU 核心数量。

  • --executor-memory 4g: 指定每个执行器的内存大小。这告诉 YARN 每个执行器可以使用的内存量。

  • /home/hadoop/demo.py: 提交的 Spark 应用程序的路径。这应该是您的 Spark 应用程序的入口点。

请确保在提交之前,Spark 相关的配置正确,并且 YARN 集群正常运行。如果有额外的依赖项,确保它们在集群中的每个节点上都可用。

代码

"""
演示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")
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54
  • 55
  • 56
  • 57
  • 58
声明:本文内容由网友自发贡献,不代表【wpsshop博客】立场,版权归原作者所有,本站不承担相应法律责任。如您发现有侵权的内容,请联系我们。转载请注明出处:https://www.wpsshop.cn/w/IT小白/article/detail/601908
推荐阅读
相关标签
  

闽ICP备14008679号