网站参数错误怎么解决方法,外包网络推广公司推广网站,cms建站系统 开源,网页布局技术有哪些目录 描述运行模式1. Windows模式代码示例 2. Local模式3. Standalone模式 RDD描述特性RDD创建代码示例#xff08;并行化创建#xff09;代码示例#xff08;读取外部数据#xff09;代码示例#xff08;读取目录下的所有文件#xff09; 算子DAGSparkSQLSparkStreaming… 目录 描述运行模式1. Windows模式代码示例 2. Local模式3. Standalone模式 RDD描述特性RDD创建代码示例并行化创建代码示例读取外部数据代码示例读取目录下的所有文件 算子DAGSparkSQLSparkStreaming 描述 Apache Spark 是用于大规模数据处理的统一分析引擎。它提供 Java、Scala、Python 和 R 中的高级 API以及支持通用执行图的优化引擎。它还支持一组丰富的高级工具包括用于 SQL 和结构化数据处理的Spark SQL 、用于机器学习的MLlib、用于图形处理的 GraphX 以及用于增量计算和流处理的结构化流。
1. Spark Core Spark的核心是Spark运行的基础。Spark Core以RDD为数据抽象提供Python、Java、Scala、R语言的API可以编程进行海量离线数据批处理计算。 2. Spark SQL Spark SQL是Spark用来操作结构化数据的组件。通过Spark SQL对数据进行处理。 3. Spark Streaming Spark Streaming是Spark平台上针对实时数据进行流式计算的组件提供了丰富的处理数据流的API。 4. Spark MLlib MLlib是Spark提供的一个机器学习算法库。MLlib不仅提供了模型评估、数据导入等额外的功能还提供了一些更底层的机器学习原语。 5. Spark GraphX GraphX是Spark面向图计算提供的框架与算法库。
运行模式
1. Windows模式
多用于本地测试不需要虚拟机或服务器。
代码示例
WordCount.scala
package com.wunaiieq//1.导入SparkConf,SparkContext
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}object WordCount {def main(args: Array[String]): Unit {//2.构建SparkConf对象并设置本地运行和程序的名称val conf new SparkConf().setMaster(local[*]).setAppName(WordCount)//3.通过SparkConf对象构建SparkContext对象val sc new SparkContext(conf)//4.读取文件并生成RDD对象val fileRdd: RDD[String] sc.textFile(data/words.txt)//5.将单词进行切割得到一个存储全部单词的集合对象val wordsRdd: RDD[String] fileRdd.flatMap(_.split( ))//6.将单词转换为Tuple2对象hello-(hello,1)val wordAndOneRdd: RDD[(String, Int)] wordsRdd.map((_, 1))//7.将元组的value按照key进行分组并对该组所有的value进行聚合操作val resultRdd: RDD[(String, Int)] wordAndOneRdd.reduceByKey(_ _)//8.通过collect方法收集RDD数据val wordCount: Array[(String, Int)] resultRdd.collect()//9.输出结果wordCount.foreach(println)}
}log4j.properties 这个没什么说的直接复制用即可
# Set everything to be logged to the console
log4j.rootCategoryERROR, console
log4j.appender.consoleorg.apache.log4j.ConsoleAppender
log4j.appender.console.targetSystem.err
log4j.appender.console.layoutorg.apache.log4j.PatternLayout
log4j.appender.console.layout.ConversionPattern%d{MM/dd HH:mm:ss} %p %c{1}: %m%n# Set the default spark-shell/spark-sql log level to WARN. When running the
# spark-shell/spark-sql, the log level for these classes is used to overwrite
# the root loggers log level, so that the user can have different defaults
# for the shell and regular Spark apps.
log4j.logger.org.apache.spark.repl.MainWARN
log4j.logger.org.apache.spark.sql.hive.thriftserver.SparkSQLCLIDriverWARN# Settings to quiet third party logs that are too verbose
log4j.logger.org.sparkproject.jettyWARN
log4j.logger.org.sparkproject.jetty.util.component.AbstractLifeCycleERROR
log4j.logger.org.apache.spark.repl.SparkIMain$exprTyperINFO
log4j.logger.org.apache.spark.repl.SparkILoop$SparkILoopInterpreterINFO
log4j.logger.org.apache.parquetERROR
log4j.logger.parquetERROR# SPARK-9183: Settings to avoid annoying messages when looking up nonexistent UDFs in SparkSQL with Hive support
log4j.logger.org.apache.hadoop.hive.metastore.RetryingHMSHandlerFATAL
log4j.logger.org.apache.hadoop.hive.ql.exec.FunctionRegistryERROR# For deploying Spark ThriftServer
# SPARK-34128Suppress undesirable TTransportException warnings involved in THRIFT-4805
log4j.appender.console.filter.1org.apache.log4j.varia.StringMatchFilter
log4j.appender.console.filter.1.StringToMatchThrift error occurred during processing of message
log4j.appender.console.filter.1.AcceptOnMatchfalse2. Local模式
一台服务器或虚拟机搞定所谓的Local模式就是不需要其他任何节点资源就可以在本地执行Spark代码的环境一般用于教学调试演示等。
# 进入spark根目录
cd /opt/module/spark/bin
# 运行视频spark-shell
./spark-shellwebUI
[atguigumaster bin]$ jps
2081 SparkSubmit
2206 Jps
[atguigumaster bin]$ netstat -anp|grep 2081
(Not all processes could be identified, non-owned process infowill not be shown, you would have to be root to see it all.)
tcp6 0 0 192.168.16.100:42050 :::* LISTEN 2081/java
tcp6 0 0 :::4040 :::* LISTEN 2081/java
tcp6 0 0 192.168.16.100:35770 :::* LISTEN 2081/java
unix 2 [ ] STREAM CONNECTED 33071 2081/java
unix 2 [ ] STREAM CONNECTED 36801 2081/java 浏览器访问
http://192.168.16.100:4040/spark-submit 以下为使用spark提交jar包示例
./spark-submit --master local[2] --class org.apache.spark.examples.SparkPi /opt/module/spark/examples/jars/spark-examples_2.12-3.1.1.jar 100
参数描述--class要执行程序的主类可以更换为自己写的应用程序的主类名称--master local[2]部署模式默认为本地模式数字 2 表示分配的虚拟 CPU 核数量spark-examples_2.12-3.2.1.jar运行的应用类所在的 jar 包实际使用时可以设定为自己打的 jar 包20程序的入口参数根据应用程序的需要可以是任何有效的输入值
几种提交方式比较
工具功能特点使用场景bin/spark-submit提交 Java/Scala/Python/R 代码到 Spark 中运行提交代码用正式场合正式提交 Spark 程序运行bin/spark-shell提供一个 Scala 解释器环境用来以 Scala 代码执行 Spark 程序解释器环境写一行执行一行测试、学习、写一行执行一行、用来验证代码等bin/pyspark提供一个 Python 解释器环境用来以 Python 代码执行 Spark 程序解释器环境写一行执行一行测试、学习、写一行执行一行、用来验证代码等
3. Standalone模式
Standalone是Spark自带的一个资源调度框架它支持完全分布式也支持HA Master角色管理整个集群的资源主要负责资源的调度和分配并进行集群的监控等职责并托管运行各个任务的Driver。如Yarn的ResourceManager。Worker角色每个从节点分配资源信息给Worker管理管理单个服务器的资源类分配对应的资源来运行Executor(Task)资源信息包含内存Memory和CPU Cores核数。如Yarn的NodeManager。Driver角色管理单个Spark任务在运行的时候的工作如Yarn的ApplicationMaster “Executor角色单个任务运行的时候的一堆工作者干活的。它是集群中工作节点Worker中的一个JVM进程负责在 Spark 作业中运行具体任务Task任务彼此之间相互独立。Spark 应用启动时Executor节点被同时启动并且始终伴随着整个 Spark应用的生命周期而存在。如果有Executor节点发生了故障或崩溃Spark应用也可以继续执行会将出错节点上的任务调度到其他Executor节点上继续运行。 Executor有两个核心功能 1.负责运行组成Spark应用的任务并将结果返回给驱动器进程。 2.它们通过自身的块管理器Block Manager为用户程序中要求缓存的 RDD 提供内存式存储。RDD 是直接缓存在Executor进程内的因此任务可以在运行时充分利用缓存数据加速运算。 总结 资源管理维度 集群资源管理者Master 单机资源管理者Worker 任务计算维度 单任务管理者Driver 单任务执行者Executor 注Executor运行于Worker进程内,由Worker提供资源供给它们运行
扩展历史服务器HistoryServer(可选)Spark Application运行完成以后保存事件日志数据至HDFS启动HistoryServer可以查看应用运行相关信息。
4. Yarn模式 Hadoop生态圈里面的一个资源调度框架Spark也是可以基于Yarn来计算的。 5. 云服务模式运行在云平台上 Kubernetes(K8S)容器模式 Spark中的各个角色运行在Kubernetes的容器内部并组成Spark集群环境。容器化部署是目前业界很流行的一项技术基于Docker镜像运行能够让用户更加方便地对应用进行管理和运维。容器管理工具中最为流行的就是(K8S)而Spark也在新版本中支持了k8s部署模式。 6. Mesos Mesos是Apache下的开源分布式资源管理框架它被称为是分布式系统的内核,在Twitter得到广泛使用,管理着Twitter超过30,0000台服务器上的应用部署但是在国内依然使用着传统的Hadoop大数据框架所以国内使用Mesos框架的并不多。
模式Spark安装机器数需启动的进程所属者应用场景Local1无Spark测试Standalone3Master及WorkerSpark单独部署Yarn1Yarn及HDFSHadoop混合部署
RDD
描述
Spark RDDResilient Distributed Dataset弹性分布式数据集代表一个不可变、可分区、元素可并行计算的集合是Spark进行数据处理的基本单元。
不可变性RDD一旦创建其数据就不可改变。对RDD的所有操作如map、filter、reduce等都会生成一个新的RDD而不会修改原始RDD。这种不可变性使得RDD在分布式计算环境下非常稳定避免了并发冲突。可分区性RDD可以分成多个分区Partition每个分区就是一个数据集片段。一个RDD的不同分区可以保存到集群中的不同节点上从而可以在集群中的不同节点上进行并行计算。分区是Spark作业并行计算的基本单位每个分区都会被一个计算任务处理分区的数量决定了并行计算的粒度。弹性RDD具有弹性容错的特点。当运算中出现异常情况导致分区数据丢失或运算失败时可以根据RDD的血统Lineage关系对数据进行重建。此外RDD的数据可以保存在内存中内存放不下时也可以保存在磁盘中实现了存储的弹性。
特性 1. 分区Partitions 含义RDD的数据被划分为多个分区每个分区是一个数据块分布在集群的不同节点上。 作用每个分区会被一个计算任务处理分区的数量决定了并行计算的粒度。用户可以在创建RDD时指定分区数如果没有指定Spark会根据集群的资源自动设置。 示例从HDFS文件创建RDD时默认分区数为文件的Block数。 2. 计算函数Compute Function 含义RDD的计算方法会作用到每个分区上。 作用当对RDD进行操作如map、filter等时Spark会对每个分区应用这个函数。 示例在map操作中计算函数会对每个元素执行指定的转换逻辑。 3. 依赖关系Dependencies 含义RDD之间存在依赖关系。 作用在部分分区数据丢失时Spark可以利用依赖关系重新计算丢失的数据而不是重新计算整个RDD提高了容错能力。 分类依赖关系分为窄依赖Narrow Dependency和宽依赖Wide Dependency。窄依赖指一个父RDD的分区最多被一个子RDD的分区使用宽依赖指一个父RDD的分区被多个子RDD的分区使用。 4. 分区器Partitioner可选只有kv型RDD才有 含义对于键值对Key-Value类型的RDD可以指定一个分区器来决定数据的分区方式。 作用分区器决定了数据在集群中的分布影响并行计算的性能。 类型Spark支持多种分区器如HashPartitioner基于哈希值分区和RangePartitioner基于范围分区。 5. 优先位置Preferred Locations可选 含义RDD分区规划应当尽量靠近数据所在的服务器 作用Spark在进行任务调度时会优先将数据分配到其存储位置进行计算减少数据传输开销提高计算效率。 示例对于HDFS文件优先位置通常是文件块所在的节点。 RDD创建
1. 通过并行化集合创建将本地集合对象转分布式RDD
val sc new SparkContext(conf)
val rdd1:RDD[Int]sc.parallelize(List(1, 2, 3, 4, 5, 6), 3)
rdd1.glom().collect()makeRdd()创建本质上也是使用sc.parallelize(…)
def makeRDD[T: ClassTag](seq: Seq[T],numSlices: Int defaultParallelism): RDD[T] withScope {parallelize(seq, numSlices)
}2. 读取外部数据源 (比如读取文件 )
//通过SparkConf对象构建SparkContext对象
val sc new SparkContext(conf)
//读取文件
val fileRdd:RDD[String] sc.textFile(data/words.txt)程序执行入口SparkContext对象 Spark RDD 编程的程序入口对象是SparkContext对象(Scala、Python、Java都是如此) 只有构建出SparkContext, 基于它才能执行后续的API调用和计算 本质上, SparkContext对编程来说, 主要功能就是创建第一个RDD出来。
代码示例并行化创建
package com.wunaiieq//1.导入SparkConf类、SparkContext
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
object CreateByParallelize {def main(args: Array[String]): Unit {//2.构建SparkConf对象。并设置本地运行和程序的名称,*表示使用全部cpu内核,可以指定数量val sparkconf new SparkConf().setMaster(local[*]).setAppName(CreateRdd1)//3.构建SparkContext对象val sparkContext new SparkContext(sparkconf)//4.通过并行化创建RDD对象将本地集合-分布式的RDD对象,如果不指定分区则根据cpu内核数进行自动分配val rdd: RDD[Int] sparkContext.parallelize(List(1, 2, 3, 4, 5, 6, 7, 8),3)//5.输出默认的分区数println(默认分区数rdd.getNumPartitions)//已经指定为3//6.collect方法将rdd对象中每个分区的数据都发送到Driver形成一个Array对象val array1: Array[Int] rdd.collect()println(rdd.collect()array1.mkString(,))//7.显示出rdd对象中元素被分布到不同分区的数据信息val array2: Array[Array[Int]] rdd.glom().collect()println(rdd.glom().collect()的内容是:)for(eleArr- array2){println(eleArr.mkString(,))}}
}代码示例读取外部数据
package com.wunaiieq//1.导入SparkConf,SparkContext类
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}object CreateByTextFile {def main(args: Array[String]): Unit {//2.构建SparkConf对象并设置本地运行和程序名val sparkConf: SparkConf new SparkConf().setMaster(local[*]).setAppName(textFile)//3.通过sparkconf创建SparkContext对象val sparkContext new SparkContext(sparkConf)//4.通过textFile读取文件//4.1.读取hdfs分布式文件系统上的文件
// val hdfsRdd: RDD[String] sparkContext.textFile(hdfs://192.168.16.100:9820/input/data.txt)
// val hdfsResult: Array[String] hdfsRdd.collect()
// println(hdfsRdd分区数hdfsRdd.getNumPartitions)
// println(hdfsRdd内容hdfsResult.mkString(,))//4.2读取本地文件val localRdd1: RDD[String] sparkContext.textFile(data/words.txt)println(localRdd1分区数localRdd1.getNumPartitions)println(localRdd1内容localRdd1.collect().mkString(,))//5.设置最小分区数val localRdd2: RDD[String] sparkContext.textFile(data/words.txt,3)println(localRdd2分区数localRdd2.getNumPartitions)println(localRdd2内容localRdd2.collect().mkString(,))//6.最小分区数设置是一个参考值Spark会有自己的判断值太大Spark不会理会val localRdd3: RDD[String] sparkContext.textFile(data/words.txt, 100)println(localRdd3的分区数localRdd3.getNumPartitions)}
}
代码示例读取目录下的所有文件
package com.wunaiieq//1.导入类
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}object CreateByWholeTextFiles {def main(args: Array[String]): Unit {//2.构建SparkConf对象并设置本地运行和程序名称val sparkConf: SparkConf new SparkConf().setMaster(local[*]).setAppName(WholeTextFiles)//3.使用sparkconf对象构建SparkContet对象val sparkContext new SparkContext(sparkConf)//5.读取指定目录下的小文件val rdd: RDD[(String, String)] sparkContext.wholeTextFiles(data)val tuples: Array[(String, String)] rdd.collect()tuples.foreach(eleprintln(ele._1,ele._2))//6.获取小文件中的内容val array: Array[String] rdd.map(_._2).collect()println(---------------------------)println(array.mkString(|))//4.关闭sparkContext对象sparkContext.stop()}
}
算子
详见如下专题RDD算子集合
DAG
详见如下专题DAG专题
SparkSQL
详见如下专题SparkSQL专题
SparkStreaming
详见如下专题SparkStreaming专题 文章转载自： http://www.morning.rzpkt.cn.gov.cn.rzpkt.cn http://www.morning.thpns.cn.gov.cn.thpns.cn http://www.morning.skql.cn.gov.cn.skql.cn http://www.morning.mqmxg.cn.gov.cn.mqmxg.cn http://www.morning.tgwfn.cn.gov.cn.tgwfn.cn http://www.morning.mkczm.cn.gov.cn.mkczm.cn http://www.morning.wqwbj.cn.gov.cn.wqwbj.cn http://www.morning.lmrcq.cn.gov.cn.lmrcq.cn http://www.morning.dbtdy.cn.gov.cn.dbtdy.cn http://www.morning.dyxzn.cn.gov.cn.dyxzn.cn http://www.morning.gmrxh.cn.gov.cn.gmrxh.cn http://www.morning.ryglh.cn.gov.cn.ryglh.cn http://www.morning.frzdt.cn.gov.cn.frzdt.cn http://www.morning.bxqry.cn.gov.cn.bxqry.cn http://www.morning.0small.cn.gov.cn.0small.cn http://www.morning.hrtwt.cn.gov.cn.hrtwt.cn http://www.morning.pumali.com.gov.cn.pumali.com http://www.morning.sdkaiyu.com.gov.cn.sdkaiyu.com http://www.morning.lyrgp.cn.gov.cn.lyrgp.cn http://www.morning.xrwbc.cn.gov.cn.xrwbc.cn http://www.morning.qzdxy.cn.gov.cn.qzdxy.cn http://www.morning.gbfuy28.cn.gov.cn.gbfuy28.cn http://www.morning.hjjkz.cn.gov.cn.hjjkz.cn http://www.morning.nwmwp.cn.gov.cn.nwmwp.cn http://www.morning.nmhpq.cn.gov.cn.nmhpq.cn http://www.morning.wxlzr.cn.gov.cn.wxlzr.cn http://www.morning.pslzp.cn.gov.cn.pslzp.cn http://www.morning.qyxwy.cn.gov.cn.qyxwy.cn http://www.morning.sqxr.cn.gov.cn.sqxr.cn http://www.morning.tzcr.cn.gov.cn.tzcr.cn http://www.morning.nfcxq.cn.gov.cn.nfcxq.cn http://www.morning.blzrj.cn.gov.cn.blzrj.cn http://www.morning.rkyw.cn.gov.cn.rkyw.cn http://www.morning.dwzwm.cn.gov.cn.dwzwm.cn http://www.morning.gychx.cn.gov.cn.gychx.cn http://www.morning.knzmb.cn.gov.cn.knzmb.cn http://www.morning.rknhd.cn.gov.cn.rknhd.cn http://www.morning.wjhnx.cn.gov.cn.wjhnx.cn http://www.morning.kdjtt.cn.gov.cn.kdjtt.cn http://www.morning.sgbsr.cn.gov.cn.sgbsr.cn http://www.morning.kxqfz.cn.gov.cn.kxqfz.cn http://www.morning.diuchai.com.gov.cn.diuchai.com http://www.morning.kwz6232.cn.gov.cn.kwz6232.cn http://www.morning.nchsz.cn.gov.cn.nchsz.cn http://www.morning.lhrwy.cn.gov.cn.lhrwy.cn http://www.morning.xdnhw.cn.gov.cn.xdnhw.cn http://www.morning.kngx.cn.gov.cn.kngx.cn http://www.morning.yckwt.cn.gov.cn.yckwt.cn http://www.morning.wanjia-sd.com.gov.cn.wanjia-sd.com http://www.morning.qszyd.cn.gov.cn.qszyd.cn http://www.morning.tsyny.cn.gov.cn.tsyny.cn http://www.morning.qtryb.cn.gov.cn.qtryb.cn http://www.morning.ljbch.cn.gov.cn.ljbch.cn http://www.morning.mfmrg.cn.gov.cn.mfmrg.cn http://www.morning.mhpmw.cn.gov.cn.mhpmw.cn http://www.morning.pwxkn.cn.gov.cn.pwxkn.cn http://www.morning.cbnxq.cn.gov.cn.cbnxq.cn http://www.morning.mczjq.cn.gov.cn.mczjq.cn http://www.morning.mxdhy.cn.gov.cn.mxdhy.cn http://www.morning.gpnwq.cn.gov.cn.gpnwq.cn http://www.morning.baohum.com.gov.cn.baohum.com http://www.morning.dgxrz.cn.gov.cn.dgxrz.cn http://www.morning.ghryk.cn.gov.cn.ghryk.cn http://www.morning.bbxbh.cn.gov.cn.bbxbh.cn http://www.morning.rjkfj.cn.gov.cn.rjkfj.cn http://www.morning.mqss.cn.gov.cn.mqss.cn http://www.morning.hrdx.cn.gov.cn.hrdx.cn http://www.morning.nwbnt.cn.gov.cn.nwbnt.cn http://www.morning.lxjcr.cn.gov.cn.lxjcr.cn http://www.morning.tdmgs.cn.gov.cn.tdmgs.cn http://www.morning.rzmlc.cn.gov.cn.rzmlc.cn http://www.morning.nbsbn.cn.gov.cn.nbsbn.cn http://www.morning.rwqj.cn.gov.cn.rwqj.cn http://www.morning.ngcth.cn.gov.cn.ngcth.cn http://www.morning.mdwtm.cn.gov.cn.mdwtm.cn http://www.morning.srbfp.cn.gov.cn.srbfp.cn http://www.morning.jmspy.cn.gov.cn.jmspy.cn http://www.morning.rxnl.cn.gov.cn.rxnl.cn http://www.morning.rjljb.cn.gov.cn.rjljb.cn http://www.morning.zhghd.cn.gov.cn.zhghd.cn
