Python大数据之PySpark(四)SparkBaseCore

news2025/1/13 9:31:33

文章目录

  • SparkBase&Core
    • 环境搭建-Spark on YARN
    • 扩展阅读-Spark关键概念
    • [了解]PySpark角色分析
    • [了解]PySpark架构
    • 后记

SparkBase&Core

  • 学习目标
  • 掌握SparkOnYarn搭建
  • 掌握RDD的基础创建及相关算子操作
  • 了解PySpark的架构及角色

环境搭建-Spark on YARN

  • Yarn 资源调度框架，提供如何基于RM，NM，Continer资源调度
  • Yarn可以替换Standalone结构中Master和Worker来使用RM和NM来申请资源

SparkOnYarn本质

  • Spark计算任务通过Yarn申请资源，SparkOnYarn
  • 将pyspark文件，经过Py4J(Python for java)转换，提交到Yarn的JVM中去运行

修改配置

  • 思考，如何搭建SparkOnYarn环境？

  • 1-需要让Spark知道Yarn(yarn-site.xml)在哪里？

  • 在哪个文件下面更改？spark-env.sh中增加YARN_CONF_DIR的配置目录

  • image-20210910103700613

  • 2-修改Yan-site.xml配置，管理内存检查,历史日志服务器等其他操作

  • 修改配置文件

  • image-20210910103802972

  • 3-需要配置历史日志服务器

  • 需要实现功能：提交到Yarn的Job可以查看19888的历史日志服务器可以跳转到18080的日志服务器上

  • 因为19888端口无法查看具体spark的executor后driver的信息，所以搭建历史日志服务器跳转

  • 3-需要准备SparkOnYarn的需要Jar包，配置在配置文件中

  • 在spark-default.conf中设置spark和yarn映射的jar包文件夹(hdfs)

  • image-20210910103918057

  • 注意，在最终执行sparkonyarn的job的时候一定重启Hadoop集群，因为更改相关yarn配置

  • 4-执行SparkOnYarn

  • 这里并不能提供交互式界面，只有spark-submit(提交任务)

  • #基于SparkOnyarn提交任务
bin/spark-submit \
--master yarn \
/export/server/spark/examples/src/main/python/pi.py  \
10

  • image-20210910103957989

小结

SparKOnYarn:使用Yarn提供了资源的调度和管理工作，真正执行计算的时候Spark本身

Master和Worker的结构是Spark Standalone结构 使用Master申请资源，真正申请到是Worker节点的Executor的Tasks线程

原来Master现在Yarn替换成ResourceManager，现在Yarn是Driver给ResourceManager申请资源

原来Worker现在Yarn替换为Nodemanager，最终提供资源的地方时hiNodeManager的Continer容器中的tasks

安装配置:

1-让spark知道yarn的位置

2-更改yarn的配置，这里需要开启历史日志服务器和管理内存检查

3-整合Spark的历史日志服务器和Hadoop的历史日志服务器，效果：通过8088的yarn的http://node1:8088/cluster跳转到18080的spark的historyserver上

4-SparkOnYarn需要将Spark的jars目录下的jar包传递到hdfs上，并且配置spark-default.conf让yarn知晓配置

5-测试，仅仅更换–master yarn

部署模式

#如果启动driver程序是在本地，称之为client客户端模式，现象：能够在client端看到结果

#如果在集群模式中的一台worker节点上启动driver，称之为cluser集群模式，现象：在client端看不到结果

  • client

>*

  • 首先 client客户端提交spark-submit任务，其中spark-submit指定–master资源，指定–deploy-mode模式

  • 由启动在client端的Driver申请资源，

  • 交由Master申请可用Worker节点的Executor中的Task线程

  • 一旦申请到Task线程，将资源列表返回到Driver端

  • Driver获取到资源后执行计算，执行完计算后结果返回到Driver端

  • 由于Drivr启动在client端的，能够直接看到结果

  • 实验：

#基于Standalone的脚本—部署模式client
#driver申请作业的资源，会向–master集群资源管理器申请
#执行计算的过程在worker中，一个worker有很多executor(进程)，一个executor下面有很多task(线程)
bin/spark-submit
–master spark://node1:7077
–deploy-mode client
–driver-memory 512m
–executor-memory 512m
/export/server/spark/examples/src/main/python/pi.py
10

  • cluster

image-20210910114736875

  • 首先 client客户端提交spark-submit任务，其中spark-submit指定–master资源，指定–deploy-mode模式

  • 由于指定cluster模式，driver启动在worker节点上

  • 由driver申请资源，由Master返回worker可用资源列表

  • 由Driver获取到资源执行后续计算

  • 执行完计算的结果返回到Driver端，

  • 由于Driver没有启动在客户端client端，在client看不到结果

  • 如何查看数据结果？

  • 需要在日志服务器上查看，演示

  • 实验：

SPARK_HOME=/export/server/spark
${SPARK_HOME}/bin/spark-submit
–master spark://node1.itcast.cn:7077,node2.itcast.cn:7077
–deploy-mode cluster
–driver-memory 512m
–executor-memory 512m
–num-executors 1
–total-executor-cores 2
–conf “spark.pyspark.driver.python=/root/anaconda3/bin/python3”
–conf “spark.pyspark.python=/root/anaconda3/bin/python3”
${SPARK_HOME}/examples/src/main/python/pi.py
10

image-20210910115446185
  • 注意事项：
  • image-20210910144911859
  • image-20210910145215725
  • 通过firstpyspark.py写的wordcount的代码，最终也是转化为spark-submit任务提交
  • 如果是spark-shell中的代码最终也会转化为spark-submit的执行脚本
  • 在Spark-Submit中可以提交driver的内存和cpu，executor的内存和cpu，–deploy-mode部署模式

Spark On Yarn两种模式

  • Spark on Yarn两种模式

  • –deploy-mode client和cluster

  • Yarn的回顾：Driver------AppMaster------RM-----NodeManager—Continer----Task

  • client模式

#deploy-mode的结构
SPARK_HOME=/export/server/spark
${SPARK_HOME}/bin/spark-submit
–master yarn
–deploy-mode client
–driver-memory 512m
–driver-cores 2
–executor-memory 512m
–executor-cores 1
–num-executors 2
–queue default
${SPARK_HOME}/examples/src/main/python/pi.py
10

#瘦身
SPARK_HOME=/export/server/spark
${SPARK_HOME}/bin/spark-submit
–master yarn
–deploy-mode client
${SPARK_HOME}/examples/src/main/python/pi.py
10

在这里插入图片描述

  • 原理：

  • image-20210910151817226

  • 1-启动Driver

  • 2-由Driver向RM申请启动APpMaster

  • 3-由RM指定NM启动AppMaster

  • 4-AppMaster应用管理器申请启动Executor(资源的封装，CPU，内存)

  • 5-由AppMaster指定启动NodeManager启动Executor

  • 6-启动Executor进程，获取任务计算所需的资源

  • 7-将获取的资源反向注册到Driver

  • 由于Driver启动在Client客户端(本地)，在Client端就可以看到结果3.1415

  • 8-Driver负责Job和Stage的划分[了解]

  • 1-执行到Action操作的时候会触发Job，不如take

  • 2-接下来通过DAGscheduler划分Job为Stages，为每个stage创建task

  • 3-接下来通过TaskScheduler将每个Stage的task分配到每个executor去执行

  • 4-结果返回到Driver端，得到结果

  • cluster：

  • 作业：

${SPARK_HOME}/bin/spark-submit
–master yarn
–deploy-mode cluster
–driver-memory 512m
–executor-memory 512m
–executor-cores 1
–num-executors 2
–queue default
–conf “spark.pyspark.driver.python=/root/anaconda3/bin/python3”
–conf “spark.pyspark.python=/root/anaconda3/bin/python3”
${SPARK_HOME}/examples/src/main/python/pi.py
10
#瘦身
${SPARK_HOME}/bin/spark-submit
–master yarn
–deploy-mode cluster
${SPARK_HOME}/examples/src/main/python/pi.py
10

image-20210910152905082>>* image-20210910152928608

  • image-20210910152952601

  • image-20210910153227501

原理：

image-20210910153937601

image-20210910154530960在这里插入图片描述

扩展阅读：两种模式详细流程

扩展阅读-Spark关键概念

扩展阅读：Spark集群角色

  • image-20210910161552868
  • Executor通过启动多个线程（task）来执行对RDD的partition进行并行计算
  • 也就是执行我们对RDD定义的例如map、flatMap、reduce等算子操作。
  • Driver：启动SparkCOntext的地方称之为Driver，Driver需要向CLusterManager申请资源，同时获取到资源后会划分Stage提交Job
  • Master：l 主要负责资源的调度和分配，并进行集群的监控等职责；
  • worker：一个是用自己的内存存储RDD的某个或某些partition；另一个是启动其他进程和线程（Executor），对RDD上的partition进行并行的处理和计算
  • Executor：一个Worker****(NodeManager)****上可以运行多个Executor，Executor通过启动多个线程（task）来执行对RDD的partition进行并行计算
  • 每个Task线程都会拉取RDD的每个分区执行计算，可以执行并行计算

扩展阅读：Spark-shell和Spark-submit

  • bin/spark-shell --master spark://node1:7077 --driver-memory 512m --executor-memory 1g

  • # SparkOnYarn组织参数

–driver-memory MEM 默认1g，Memory for driver (e.g. 1000M, 2G) (Default: 1024M). Driver端的内存

–driver-cores NUM 默认1个，Number of cores used by the driver, only in cluster mode(Default: 1).

–num-executors NUM 默认为2个，启动多少个executors

–executor-cores NUM 默认1个，Number of cores used by each executor，每个executou需要多少cpucores

–executor-memory 默认1G，Memory per executor (e.g. 1000M, 2G) (Default: 1G) ，每个executour的内存

–queue QUEUE_NAME The YARN queue to submit to (Default: “default”).

bin/spark-submit --master yarn \

–deploy-mode cluster \

–driver-memory 1g \

–driver-cores 2 \

–executor-cores 4 \

–executor-memory 512m \

–num-executors 10 \

path/XXXXX.py \

10

扩展阅读：命令参数

–driver-memory MEM 默认1g，Memory for driver (e.g. 1000M, 2G) (Default: 1024M). Driver端的内存

–driver-cores NUM 默认1个，Number of cores used by the driver, only in cluster mode(Default: 1).

–num-executors NUM 默认为2个，启动多少个executors

–executor-cores NUM 默认1个，Number of cores used by each executor，每个executou需要多少cpucores

–executor-memory 默认1G，Memory per executor (e.g. 1000M, 2G) (Default: 1G) ，每个executour的内存

–queue QUEUE_NAME The YARN queue to submit to (Default: “default”).

MAIN函数代码执行

  • image-20210910165027137
  • Driver端负责申请资源包括关闭资源，负责任务的Stage的切分
  • Executor执行任务的计算
  • 一个Spark的Application有很多Job
  • 一个Job下面有很多Stage
  • 一个Stage有很多taskset
  • 一个Taskset有很多task任务构成的额
  • 一个rdd分task分区任务都需要executor的task线程执行计算

再续 Spark 应用

[了解]PySpark角色分析

  • Spark的任务执行的流程
  • 面试的时候按照Spark完整的流程执行即可
  • image-20210910175410393
  • Py4J–Python For Java–可以在Python中调用Java的方法
  • 因为Python作为顶层的语言，作为API完成Spark计算任务，底层实质上还是Scala语言调用的
  • 底层有Python的SparkContext转化为Scala版本的SparkContext
  • ****为了能在Executor端运行用户定义的Python函数或Lambda表达****式，则需要为每个Task单独启一个Python进程，通过socket通信方式将Python函数或Lambda表达式发给Python进程执行。

[了解]PySpark架构

  • image-20210910180018622

后记

📢博客主页：https://manor.blog.csdn.net

📢欢迎点赞 👍 收藏 ⭐留言 📝 如有错误敬请指正！
📢本文由 Maynor 原创，首发于 CSDN博客🙉
📢感觉这辈子，最深情绵长的注视，都给了手机⭐
📢专栏持续更新,欢迎订阅：https://blog.csdn.net/xianyu120/category_12453356.html

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

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈，一经查实，立即删除！

相关文章

UE中制作棋盘格材质效果

UE中制作棋盘格材质效果

在UE中通过这个小技巧制作棋盘格材质效果，可以快速预览UV拉伸情况，方便调试导入的模型。 1.操作步骤 1.1 首先新建材质，Shading Model（着色模式）设置为Unlit（无光照）： 1.2 我们…
阅读更多...
管道-匿名管道

管道-匿名管道

一、管道介绍 管道（Pipe）是一种在UNIX和类UNIX系统中用于进程间通信的机制。它允许一个进程的输出直接成为另一个进程的输入，从而实现数据的流动。管道是一种轻量级的通信方式，用于协调不同进程的工作。 1. 创建和使用管道&#…
阅读更多...
Docker Alist 在线网盘部署

Docker Alist 在线网盘部署

文章目录 拉取镜像创建并运行查看容器自动生成的密码在浏览器中进行访问 挂载本地磁盘 拉取镜像 docker pull xhofe/alist-aria2创建并运行 # -v /data/alist:/opt/alist/data 挂载本地目录 docker run -d --restartalways -v /data/alist:/opt/alist/data -p 5244:5244 -e P…
阅读更多...
国庆作业6

国庆作业6

TCP服务器 #include "head.h" #define PORT 2580 //端口号 #define IP "192.168.31.219" //本机IP int main(int argc, const char *argv[]) {sqlite3* dbNULL;if(sqlite3_open("./my.db",&db)!SQLITE_OK){fprintf(stde…
阅读更多...
FFmpeg 命令：从入门到精通 | ffmpeg filter（过滤器 / 滤镜）

FFmpeg 命令：从入门到精通 | ffmpeg filter（过滤器 / 滤镜）

FFmpeg 命令：从入门到精通 | ffmpeg filter（过滤器 / 滤镜） FFmpeg 命令：从入门到精通 | ffmpeg filter（过滤器 / 滤镜）ffmpeg fliter 基本内置变量视频裁剪文字水印图片水印画中画视频多宫格处理 FFmpeg 命…
阅读更多...
BIOS 如何确定引导扇区的位置

BIOS 如何确定引导扇区的位置

机械硬盘的信息都是存储在磁盘的表面，他们是由一圈一圈的磁道组成的。以前的磁盘的不同磁道的扇区数量是相同的，这就导致了很大的浪费，因为最外面的磁道的密度是最低的，信息是最安全的，所以最外面的磁道是0磁道。后来由…
阅读更多...
NPDP产品经理知识(产品创新管理)

NPDP产品经理知识(产品创新管理)

复习文化，团队与领导力 产品创新管理： 如何树立愿景： 如何实现产品战略 计划 实施产品开发： 商业化，营销计划，推广活动 管理产品生命周期： 新式走向市场的流程：
阅读更多...
win10 关闭病毒防护

win10 关闭病毒防护

windows10彻底关闭Windows Defender的4种方法 - 知乎
阅读更多...
科技+智慧+颜值，智慧公厕黑科技提升城市形象

科技+智慧+颜值，智慧公厕黑科技提升城市形象

现代社会的科技和智慧正以惊人的速度渗透到我们生活的各个方面，包括公共设施。而作为城市形象的重要组成部分之一，公厕也在不断创新中迎来了智慧的时代。 在传统的公厕中，一些问题一直困扰着我们，比如厕位的占用情况、空气质量的…
阅读更多...
腾讯云域名API解析升级版本（通过Java实现）腾讯云动态公网IP绑定域名实现内网服务器公网穿透

腾讯云域名API解析升级版本（通过Java实现）腾讯云动态公网IP绑定域名实现内网服务器公网穿透

公众号推广： 目前CSDN进行VIP可见，文章可在微信公众号进行免费的阅读。 文章内容经过认证实践，比较的清晰易懂，适合初次接触的人员。 请关注微信公众号：菜鸟编程踩坑之路，进入公众号搜索关键词 内网穿透 需求场景： 首先我自己组装了一台自己的服务器，相比较购买的阿…
阅读更多...
微信小程序——CSS3渐变

微信小程序——CSS3渐变

SS3 渐变（gradients）可以在两个或多个指定的颜色之间显示平稳的过渡。CSS3 定义了两种类型的渐变（gradients）： 说明 1、线性渐变（Linear Gradients）- 向下/向上/向左/向右/对角方向&#xff1…
阅读更多...
基于体素场景的摄像机穿模处理

基于体素场景的摄像机穿模处理

基于上一篇一种基于体素的射线检测 使用射线处理第三人称摄像头穿模问题 基于体素的第三人称摄像机拉近简单处理 摄像机移动至碰撞点处 简单的从角色身上发射一条射线到摄像机，中途遇到碰撞就把摄像机移动至该碰撞点 public void UpdateDistance(float defaultDist…
阅读更多...
机器学习（监督学习）笔记

机器学习（监督学习）笔记

笔记内容 代码部分 # 实验2-1 # 批梯度下降 import pandas as pd import numpy as np import random as rd import matplotlib.pyplot as plt # load dataset df pd.read_csv(temperature_dataset.csv) data np.array(df) y0 np.array([i[0] for i in data]) # 第一列作为…
阅读更多...
Docker项目部署

Docker项目部署

目录 一、前端项目部署 1、上传文件 2、开启容器 3、测试 二、后端项目部署 1、打包java项目 2、将jar包和Dockerfile文件长传到Linux系统 3、构建镜像 4、开启容器 5、测试 一、前端项目部署 1、上传文件 里面包括页面和配置文件 worker_processes 1;events {worker…
阅读更多...
云畅科技TMS解决方案助力华菱线缆实现智能货运管理

云畅科技TMS解决方案助力华菱线缆实现智能货运管理

9月26日下午，湖南华菱线缆股份有限公司TMS物流系统上线启动会成功举办，由云畅科技倾力打造的华菱线缆TMS物流系统正式上线运行，标志着湖南华菱线缆股份有限公司在智能化物流货运管理领域的一次重大突破。 湖南华菱线缆股份有限公司董事兼总经…
阅读更多...
C#捕捉全局异常

C#捕捉全局异常

1.运行图片 2.源码 using System; using System.Collections.Generic; using System.Linq; using System.Threading.Tasks; using System.Windows.Forms;namespace 捕捉全局异常 {internal static class Program{/// <summary>/// 应用程序的主入口点。/// </summary…
阅读更多...
竞赛选题 机器视觉opencv答题卡识别系统

竞赛选题 机器视觉opencv答题卡识别系统

0 前言 &#x1f525; 优质竞赛项目系列&#xff0c;今天要分享的是 &#x1f6a9; 答题卡识别系统 - opencv python 图像识别 该项目较为新颖&#xff0c;适合作为竞赛课题方向&#xff0c;学长非常推荐&#xff01; &#x1f947;学长这里给一个题目综合评分(每项满分5分…
阅读更多...
【Spring】更简单的读取和存储对象

【Spring】更简单的读取和存储对象

更简单的读取和存储对象 一. 存储 Bean 对象1. 前置工作&#xff1a;配置扫描路径2. 添加注解存储 Bean 对象Controller&#xff08;控制器存储&#xff09;Service&#xff08;服务存储&#xff09;Repository&#xff08;仓库存储&#xff09;Component&#xff08;组件存储&…
阅读更多...
基于transformer的心脑血管心脏病疾病预测

基于transformer的心脑血管心脏病疾病预测

视频讲解：基于transformer的心脑血管疾病预测 完整数据代码分享_哔哩哔哩_bilibili 数据展示： 完整代码： # pip install openpyxl -i https://pypi.tuna.tsinghua.edu.cn/simple/ # pip install optuna -i https://pypi.tuna.tsinghua.edu.cn/simple/ import numpy as np …
阅读更多...
C++ - 位图 - bitset 容器介绍

C++ - 位图 - bitset 容器介绍

前言 之前的两篇博客已经完成 闭散列的开放地址法的哈希表 和 哈希桶基本实现&#xff0c;和对 unordered_set 和 unordered_map 的封装 &#xff1a;C - 封装 unordered_set 和 unordered_map - 哈希桶的迭代器实现_chihiro1122的博客-CSDN博客C - 开散列的拉链法&#xff08…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023