Spark核心名词解释与编程

2024-04-29 06:36

本文主要是介绍Spark核心名词解释与编程,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

Spark核心概念

名词解释

1)ClusterManager:在Standalone(上述安装的模式,也就是依托于spark集群本身)模式中即为Master(主节点),控制整个集群,监控Worker。在YARN模式中为资源管理器ResourceManager(国内spark主要基于yarn集群运行,欧美主要基于mesos来运行)。

2)Application:Spark的应用程序,包含一个Driver program和若干Executor。

3)SparkConf:负责存储配置信息。作用相当于hadoop中的Configuration。

4)SparkContext:Spark应用程序的入口,负责调度各个运算资源,协调各个Worker Node上的Executor。

5)Worker:从节点,负责控制计算节点,启动Executor。在YARN模式中为NodeManager,负责计算节点的控制,启动的进程叫Container。

6)Driver:运行Application的main()函数并创建SparkContext(是spark中最重要的一个概念,是spark编程的入口,作用相当于mr中的Job)。

7)Executor:执行器,在worker node上执行任务的组件、用于启动线程池运行任务。每个Application拥有独立的一组Executors。

8)RDD:Spark的基本计算单元,一组RDD可形成执行的有向无环图RDD Graph。

9)RDD是弹性式分布式数据集,理解从3个方面去说:弹性、数据集、分布式。是Spark的第一代的编程模型。

10)DAGScheduler:实现将Spark作业分解成一到多个Stage,每个Stage根据RDD的Partition个数决定Task的个数,然后生成相应的Taskset放到TaskScheduler中。DAGScheduler就是Spark的大脑,中枢神经。

11)TaskScheduler:将任务(Task)分发给Executor执行。

12)Stage:一个Spark作业一般包含一到多个Stage。

13)Task:一个Stage包含一到多个Task,通过多个Task实现并行运行的功能。task的个数由rdd的partition分区决定,spark是一个分布式计算程序,所以一个大的计算任务,就会被拆分成多个小的部分,同时进行计算。一个partition对应一个task任务。

14)Transformations:转换(Transformations) (如:map, filter, groupBy, join等),Transformations操作是Lazy的,也就是说从一个RDD转换生成另一个RDD的操作不是马上执行,Spark在遇到Transformations操作时只会记录需要这样的操作,并不会去执行,需要等到有Actions操作的时候才会真正启动计算过程进行计算。

15)Actions:操作/行动(Actions)算子 (如:count, collect, foreach等),Actions操作会返回结果或把RDD数据写到存储系统中。Actions是触发Spark启动计算的动因。

Spark官网组件说明

官网组件说明如图-18所示:

图-18 Spark组件通信架构图

Spark应用程序作为集群上的独立进程集运行,由主程序(称为驱动程序)中的SparkContext对象协调。

具体来说,要在集群上运行,SparkContext可以连接到几种类型的集群管理器(Spark自己的独立集群管理器、Mesos或YARN),这些管理器可以跨应用程序分配资源。一旦连接,Spark将获取集群中节点上的执行器,这些执行器是为应用程序运行计算和存储数据的进程。接下来,它将应用程序代码(由传递给SparkContext的JAR或Python文件定义)发送给执行器。最后,SparkContext将任务发送给执行器以运行。

Spark编程体验

项目依赖管理

<dependencies><dependency><groupId>org.scala-lang</groupId><artifactId>scala-library</artifactId><version>2.12.10</version></dependency><dependency><groupId>org.apache.spark</groupId><artifactId>spark-core_2.12</artifactId><version>3.2.1</version></dependency><dependency><groupId>org.apache.spark</groupId><artifactId>spark-sql_2.12</artifactId><version>3.2.1</version></dependency><dependency><groupId>mysql</groupId><artifactId>mysql-connector-java</artifactId><version>8.0.23</version></dependency><dependency><groupId>org.apache.spark</groupId><artifactId>spark-hive_2.12</artifactId><version>3.2.1</version></dependency><dependency><groupId>org.apache.spark</groupId><artifactId>spark-streaming_2.12</artifactId><version>3.2.1</version></dependency><dependency><groupId>org.apache.spark</groupId><artifactId>spark-streaming-kafka-0-10_2.12</artifactId><version>3.2.1</version></dependency>
</dependencies><build><finalName>chapter1.WordCount</finalName><plugins><plugin><groupId>net.alchim31.maven</groupId><artifactId>scala-maven-plugin</artifactId><version>3.4.6</version><executions><execution><goals><goal>compile</goal><goal>testCompile</goal></goals></execution></executions></plugin></plugins>
</build>

项目编码

spark入门程序wordcount:

package com.fesco.bigdata.spark
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
/*** scala版本的wordcount*/
object ScalaWordCountApp {
def main(args: Array[String]): Unit = {
val conf = new SparkConf()
.setAppName(s"${ScalaWordCountApp.getClass.getSimpleName}")
.setMaster("local[*]")
val sc = new SparkContext(conf)
//加载数据
val file: RDD[String] = sc.textFile("file:/E:/data/spark/hello.txt")//按照分隔符进行切分
val words:RDD[String] = lines.flatMap(line => line.split("\\s+"))//每个单词记为1次
val pairs:RDD[(String, Int)] = words.map(word => (word, 1))//聚合数据
val ret:RDD[(String, Int)] = pairs.reduceByKey(myReduceFunc)
//export data to external system
ret.foreach(println)}
sc.stop()
}
def myReduceFunc(v1: Int, v2: Int): Int = {
v1 + v2
}
}

Master URL说明

首先在编程过程中,至少需要给spark程序传递一个参数master-url,通过sparkConf.setMaster来完成。改参数,代表的是spark作业的执行方式,或者指定的spark程序的cluster-manager的类型。

表-1 模式选择

master

含义

local

程序在本地运行,同时为本地程序提供一个线程来处理

local[M]

程序在本地运行,同时为本地程序分配M个工作线程

来处理

local[*]

程序在本地运行,同时为本地程序分配机器可用的CPU core的个数工作线程来处理

local[M, N]

程序在本地运行,同时为本地程序分配M个工作线程来处理,如果提交程序失败,会进行最多N次的重试

spark://ip:port

基于standalone的模式运行,提交撑到ip对应的master上运行

spark://ip1:port1,ip2:port2

基于standalone的ha模式运行,提交撑到ip对应的master上运行

yarn/启动脚本中的deploy-mode配置为cluster

基于yarn模式的cluster方式运行,SparkContext的创建在NodeManager上面,在yarn集群中

yarn/启动脚本中的deploy-mode配置为client

基于yarn模式的client方式运行,SparkContext的创建在提交程序的那台机器上面,不在yarn集群中

spark程序的其他提交方式

加载hdfs中的文件:

object RemoteSparkWordCountOps {def main(args: Array[String]): Unit = {//创建程序入口val conf = new SparkConf().setAppName("wc").setMaster("local[*]")val sc = new SparkContext(conf)//设置日志级别sc.setLogLevel("WARN")//加载数据val file = sc.textFile("hdfs://hadoop101:8020//wordcount//words.txt")//切分val spliFile: RDD[String] = file.flatMap(_.split(" "))//每个单词记为1次val wordAndOne: RDD[(String, Int)] = spliFile.map((_, 1))//聚合val wordAndCount: RDD[(String, Int)] = wordAndOne.reduceByKey(_ + _)//打印输出wordAndCount.foreach(println)//释放资源sc.stop()
}}

提交spark程序到集群中

首先需要将spark-core模块进行打包,其次上传到集群中,才可以进行提交作业到spark或者yarn集群中运行。

1)Client:

bin/spark-submit \--class chapter1.WordCount \--master spark://hadoop101:7077 \/root/word.jar \hdfs://hadoop101:8020/wordcount/words.txt

2)Cluster:

bin/spark-submit \--class chapter1.WordCount \--master spark://hadoop101:7077 \/root/word.jar \hdfs://hadoop101:8020/wordcount/words.txt \hdfs://hadoop101:8020/wordcount/output1

这篇关于Spark核心名词解释与编程的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/945346

相关文章

揭秘Python Socket网络编程的7种硬核用法

《揭秘PythonSocket网络编程的7种硬核用法》Socket不仅能做聊天室,还能干一大堆硬核操作,这篇文章就带大家看看Python网络编程的7种超实用玩法,感兴趣的小伙伴可以跟随小编一起... 目录1.端口扫描器:探测开放端口2.简易 HTTP 服务器:10 秒搭个网页3.局域网游戏:多人联机对战4.

Java并发编程必备之Synchronized关键字深入解析

《Java并发编程必备之Synchronized关键字深入解析》本文我们深入探索了Java中的Synchronized关键字,包括其互斥性和可重入性的特性,文章详细介绍了Synchronized的三种... 目录一、前言二、Synchronized关键字2.1 Synchronized的特性1. 互斥2.

Python异步编程中asyncio.gather的并发控制详解

《Python异步编程中asyncio.gather的并发控制详解》在Python异步编程生态中,asyncio.gather是并发任务调度的核心工具,本文将通过实际场景和代码示例,展示如何结合信号量... 目录一、asyncio.gather的原始行为解析二、信号量控制法:给并发装上"节流阀"三、进阶控制

Linux find 命令完全指南及核心用法

《Linuxfind命令完全指南及核心用法》find是Linux系统最强大的文件搜索工具,支持嵌套遍历、条件筛选、执行动作,下面给大家介绍Linuxfind命令完全指南,感兴趣的朋友一起看看吧... 目录一、基础搜索模式1. 按文件名搜索(精确/模糊匹配)2. 排除指定目录/文件二、根据文件类型筛选三、时间

C#多线程编程中导致死锁的常见陷阱和避免方法

《C#多线程编程中导致死锁的常见陷阱和避免方法》在C#多线程编程中,死锁(Deadlock)是一种常见的、令人头疼的错误,死锁通常发生在多个线程试图获取多个资源的锁时,导致相互等待对方释放资源,最终形... 目录引言1. 什么是死锁?死锁的典型条件:2. 导致死锁的常见原因2.1 锁的顺序问题错误示例:不同

PyCharm接入DeepSeek实现AI编程的操作流程

《PyCharm接入DeepSeek实现AI编程的操作流程》DeepSeek是一家专注于人工智能技术研发的公司,致力于开发高性能、低成本的AI模型,接下来,我们把DeepSeek接入到PyCharm中... 目录引言效果演示创建API key在PyCharm中下载Continue插件配置Continue引言

SpringBoot操作spark处理hdfs文件的操作方法

《SpringBoot操作spark处理hdfs文件的操作方法》本文介绍了如何使用SpringBoot操作Spark处理HDFS文件,包括导入依赖、配置Spark信息、编写Controller和Ser... 目录SpringBoot操作spark处理hdfs文件1、导入依赖2、配置spark信息3、cont

C#反射编程之GetConstructor()方法解读

《C#反射编程之GetConstructor()方法解读》C#中Type类的GetConstructor()方法用于获取指定类型的构造函数,该方法有多个重载版本,可以根据不同的参数获取不同特性的构造函... 目录C# GetConstructor()方法有4个重载以GetConstructor(Type[]

Andrej Karpathy最新采访:认知核心模型10亿参数就够了,AI会打破教育不公的僵局

夕小瑶科技说 原创  作者 | 海野 AI圈子的红人,AI大神Andrej Karpathy,曾是OpenAI联合创始人之一,特斯拉AI总监。上一次的动态是官宣创办一家名为 Eureka Labs 的人工智能+教育公司 ,宣布将长期致力于AI原生教育。 近日,Andrej Karpathy接受了No Priors(投资博客)的采访,与硅谷知名投资人 Sara Guo 和 Elad G

Linux 网络编程 --- 应用层

一、自定义协议和序列化反序列化 代码: 序列化反序列化实现网络版本计算器 二、HTTP协议 1、谈两个简单的预备知识 https://www.baidu.com/ --- 域名 --- 域名解析 --- IP地址 http的端口号为80端口,https的端口号为443 url为统一资源定位符。CSDNhttps://mp.csdn.net/mp_blog/creation/editor