Python大数据之PySpark(四)SparkBaseCore

文章目录

  • 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.rhkb.cn/news/147308.html

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

相关文章

Linux 下如何调试代码

Linux 下如何调试代码

debug 和 release 在Linux下的默认模式是什么？ 是release模式 那你怎么证明他就是release版本? 我们知道如果一个程序可以被调试，那么它一定是debug版本，如果它是release版本，它是没法被调试的，所以说我们可以来调试一…
阅读更多...
FPGA project : TFT_LCD

FPGA project : TFT_LCD

实验目标： 驱动TFT_LCD显示十色彩条。 重点掌握的知识： 1，液晶显示器，简称LCD(Liquid Crystal Display)，相对于上一代CRT显示器(阴极射线管显示器)，LCD显示器具有功耗低、体积小、承载的信息量大及不伤眼…
阅读更多...
王道考研操作系统——I/O管理

王道考研操作系统——I/O管理

I/O设备的基本概念 键盘：输入设备（把设备准备好的数据读入计算机当中）； 显示器：输出设备（把计算机中准备好的数据写出到设备上）； 移动硬盘：既是输入又是输出 中断驱动…
阅读更多...
数据挖掘（2）数据预处理

数据挖掘（2）数据预处理

一、数据预处理 1.1概述 数据预处理的重要性 杂乱性：如命名规则。重复性：同一客观事再不完整性：噪声数据：数据中存在错误或异常的现象。 数据预处理的常见方法 数据清洗：去掉数据中的噪声，纠正不一致。数…
阅读更多...
HTML的学习 Day02（列表、表格、表单）

HTML的学习 Day02（列表、表格、表单）

文章目录 一、列表列表主要分为以下三种类型：1. 无序列表（Unordered List）：2. 有序列表（Ordered List）：将有序列表的数字改为字母或自定义内容li.../li 列表项标签中value属性，制定列…
阅读更多...
OpenCV实现视频的追踪（meanshift、Camshift）

OpenCV实现视频的追踪（meanshift、Camshift）

目录 1，meanshift 1.1 算法流程 1.2 算法实现 1.3 代码实现 1.4 结果展示 1，meanshift 1.1 算法流程 1.2 算法实现 1.3 代码实现 import numpy as np import cv2 as cv# 读取视频 cap cv.VideoCapture(video.mp4)# 检查视频是否成功打开 if n…
阅读更多...
分布式应用程序协调服务 ZooKeeper 详解

分布式应用程序协调服务 ZooKeeper 详解

目录 1、ZooKeeper简介 2、ZooKeeper的使用场景 3、ZooKeeper设计目的 4、ZooKeeper数据模型 5、ZooKeeper几个重要概念 5.1、ZooKeeper Session 5.2、ZooKeeper Watch 5.3、Consistency Guarantees 6、ZooKeeper的工作原理 6.1、Leader Election 6.2、Leader工作流…
阅读更多...
NPDP产品经理知识(产品创新管理)

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

复习文化，团队与领导力 产品创新管理： 如何树立愿景： 如何实现产品战略 计划 实施产品开发： 商业化，营销计划，推广活动 管理产品生命周期： 新式走向市场的流程：
阅读更多...
【Docker】docker拉取镜像错误 missing signature key

【Docker】docker拉取镜像错误 missing signature key

问题 当我使用docker拉取一个特定的镜像时，提示错误： 错误 missing signature key 但是拉取其他镜像又可以访问，，，，于是，我怀疑是否是docker版本问题。 docker --version结果确实&#xff0…
阅读更多...
操作系统原理-习题汇总

操作系统原理-习题汇总

临近毕业，整理一下过去各科习题及资料等，以下为操作系统原理的习题汇总，若需要查找题目，推荐CtrlF或commandF进行全篇快捷查找。 操作系统原理 作业第一次作业选择题简答题 第二次作业选择题简答题 第三次作业选择题简答题 第四次…
阅读更多...
ctfshow—1024系列练习

ctfshow—1024系列练习

1024 柏拉图 有点像rce远程执行，有四个按钮，分别对应四份php文件，开始搞一下。一开始，先要试探出 文件上传到哪里？ 怎么读取上传的文件？ 第一步：试探上传文件位置 直接用burp抓包，…
阅读更多...
凉鞋的 Godot 笔记 105. 第一个通识：编辑-测试 循环

凉鞋的 Godot 笔记 105. 第一个通识：编辑-测试 循环

105. 第一个通识：编辑-测试 循环 在这一篇，我们简单聊聊此教程中所涉及的一个非常重要的概念：循环。 我们在做任何事情都离不开某种循环，比如每天的 24 小时循环，一日三餐循环，清醒-睡觉循环。 在学习一…
阅读更多...
通过java向jar写入新文件

通过java向jar写入新文件

文章目录 原始需求分析实施步骤引入依赖核心编码运行效果 原始需求 有网友提问： 我想在程序中动态地向同一个jar包中添加文件，比如，我的可执行jar包是test.jar,我要在它运行时生成一些xml文件并将这些文件添加到test.jar中,请问如何实现&…
阅读更多...
C#制做一个 winform下的表情选择窗口

C#制做一个 winform下的表情选择窗口

能力有限，别人可能都是通过其他方式实现的，我这里简单粗暴一些，直接通过点击按钮后弹出个新窗体来实现。 1、先在form1上增加一个toolstrip控件，再增加个toolstripbutton按钮，用来点击后弹出新窗体，如图&a…
阅读更多...
centos 部署nginx 并配置https

centos 部署nginx 并配置https

centos版本：centos 7.8 （最好不要用8，8的很多用法和7相差很大） 一.安装nginx 1。下载Nginx安装包：首先，访问Nginx的官方网站（https://nginx.org/）或您选择的镜像站点，找…
阅读更多...
阿里云ACP知识点（三）

阿里云ACP知识点（三）

1、弹性伸缩不仅提供了在业务需求高峰或低谷时自动调节ECS实例数量的能力，而且提供了ECS实例上自动部署应用的能力。弹性伸缩的伸缩配置支持多种特性，例如______,帮助您高效、灵活地自定义ECS实例配置，满足业务需求。 标签、密钥对、 实例RAM…
阅读更多...
AWS-Lambda之导入自定义包-pip包

AWS-Lambda之导入自定义包-pip包

参考文档： https://repost.aws/zh-Hans/knowledge-center/lambda-import-module-error-python https://blog.csdn.net/fxtxz2/article/details/112035627 简单来说,以 " alibabacloud_dyvmsapi20170525 " 包为例 ## 创建临时目录 mkdir /tmp cd ./tmp …
阅读更多...
机器学习（监督学习）笔记

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

笔记内容 代码部分 # 实验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]) # 第一列作为…
阅读更多...
QT、C++实现地图导航系统（mapSystem）

QT、C++实现地图导航系统（mapSystem）

文章目录 地图导航系统项目应用背景技术栈选择数据处理算法实现界面实现源码展示成果展示源码下载 （免费） 地图导航系统 项目应用背景 电子地图导航系统的主要目的是为用户提供精确、实时的导航和位置信息，以帮助他们在城市或地区内轻松找到…
阅读更多...
固定式工业RFID读写器有哪些特点?如何选型？

固定式工业RFID读写器有哪些特点?如何选型？

工业读写器可分为便携式读写器和固定式读写器，固定式读写器主要是将读写器固定在某一位置钟，常常应用在工位、生产、进入库等场景中。但是很多人并不了解固定式读写器都有哪些特点，具体应该怎么选型，下面我们就一起来了解一下。 固…
阅读更多...
最新文章
推荐文章

Copyright @ 2022~2023

友情链接： 我的编程经验分享 一条长河网 尧图网站开发