【spark你妈喊你回家吃饭-05】RDD编程之旅基础篇-01
2017-06-16 17:54
519 查看
1.RDD工作流程
1.1 RDD理解
RDD是spark特有的数据模型,谈到RDD就会提到什么弹性分布式数据集,什么有向无环图,本文暂时不去展开这些高深概念,在阅读本文时候,大家可以就把RDD当作一个数组,这样的理解对我们学习RDD的API是非常有帮助的。本文所有示例代码都是使用scala语言编写的。RDD的执行过程如下:从外部数据创建出输入RDD,或者从驱动程序分发驱动程序中的对象集合
对RDD进行转化,一个RDD转化为一个新的RDD,如filter()操作
如果需要重用,告知RDD执行persist()操作
执行action触发计算并行计算,spark先优化再执行计算,如count()和first()
RDD的创建有2种方式
1)从驱动程序分发驱动程序中的对象集合
从内存里构造RDD,使用的方法:makeRDD和parallelize方法
val rdd01 = sc.makeRDD(List(1,2,3,4,5,6)); val r01 = rdd01.map { x => x * x } println(r01.collect().mkString(",")) /* Array */ val rdd02 = sc.makeRDD(Array(1,2,3,4,5,6)) val r02 = rdd02.filter { x => x < 5} println(r02.collect().mkString(",")) val rdd03 = sc.parallelize(List(1,2,3,4,5,6), 1) val r03 = rdd03.map { x => x + 1 } println(r03.collect().mkString(",")) /* Array */ val rdd04 = sc.parallelize(Array(1,2,3,4,5,6) ,1) val r04 = rdd04.filter { x => x > 3 } println(r04.collect().mkString(","))
1.2.makeRDD和parallelize的区别
makeRDD有两种实现方式,第一种方式parallelize声明都一样,接收的参数和parallelize完全一样,def makeRDD[T:ClassTag],这种实现方式的makeRDD依赖了parallelize;makeRDD第二种实现方式defmakeRDD[T:ClassTag](T,Seq(String)))第一种:mkRDD实现方式
val blog1=sc.parallelize(List(1,2,3)); val blog2=sc.makeRDD(List(1,2,3)
第二种:mkRDD实现方式
val seq=List((1,List("a","b","c")),(2,List("aa","bb","cc")));
val seq=List((1,List("a","b","c")),(2,List("aa","bb","cc"))); val blog3=sc.makeRDD(seq); blog3.preferredLocations(blog3.partitions(0)); blog3.preferredLocations(blog3.partitions(1));
2.[b]WordCount解说[/b]
WordCount是分布式编程的入门示例,本节也从WordCount举例说明RDD DEMO2.1.Spark shell版本
//加载hdfs上的文件 val txtFile ="/tmp/test/core-site.xml" ; val txtData = sc.textFile(txtFile); //将上一步生成的RDD对象保存到缓存中,在此之后Spark就不需要在每次数据查询时都重新计算 txtData.cache() ; // flatMap先映射后扁平化, val wcData = txtData.flatMap(l =>l.split(" ")).map(word => (word, 1)).reduceByKey(_ + _); //可以提取出所有rdd里的数据项,逐行输出 wcData.collect().foreach(println);
备注:
A. 关于spark-shell的启动参数指定
bin/spark-shell --executor-memory 1G --total-executor-cores10 --executor-cores 1 --master yarn-client --driver-class-path /usr/local/tdr_hadoop/spark/spark-1.6.0-bin-hadoop2.6/lib/mysql-connector-java-5.1.40-bin.jar
--executor-memory: 指定每个executor(执行器)占用的内存
--total-executor-cores: 所有executor总共使用的cpu核数
--executor-cores:每个executor使用的cpu核数
--driver-class-path:指定要加载的jar包
--master:
local[8]:表示在本地运行,数据会下载到接口机本地来执行,单机版
spark://master01:7077:表示在集群上运行应用程序,指定任务提交的集群路径在哪里。这就需要提前启动一个真实的Standalone集群。可以指定多个master的地址,用逗号隔开。
yarn-client:在客户模式上,driver与提交程序的客户端在一个进程
yarn-cluster:在集群模式上,driver是从集群中的一个worker进程中启动的,这个进程只要完成了提交作业任务就会退出,不会等待提交的应用程序的完成。Spark-shell时,必须使用yarn-client模式,因为你要在client上写SQL。
B.spark-shell 是一个spark application,运行时需要向资源管理器申请资源,如standalone spark、YARN、Mesos。本例向standalone spark申请资源,所以在运行spark-shell时需要指向申请资源的standalonespark集群信息,其参数为MASTER。
如果未在spark-env.sh中申明MASTER,则使用命令MASTER=spark://cdh1:7077bin/spark-shell启动;如果已经在spark-env.sh中申明MASTER,则可以直接用bin/spark-shell启动。
由于spark-shell缺省的情况下,会申请所有的CPU资源
2.2. java 版本
搭建Spark开发环境(1)前提:配置好jdk和scala到windows
(2)安装Intellij去官网下载Intellij:https://www.jetbrains.com/idea/,在windows环境下双击安装即可
(3)安装scala插件,安装好scala插件后,点击restart重启intellij
(4)、使用Intellij写WordCount代码
a.新建scala工程:File -> new -> project -> scala project –>scala,项目名称:spark02
在src目录下,建立cn.com包,在该包下建立object 类,命名为word,完成word.scala代码如下所示:
import org.apache.spark.SparkConf import org.apache.spark.SparkContext import org.apache.spark.SparkContext._ /** * Created by Administrator on 2016/11/2. */ object word { def main(args: Array[String]) { if(args.length < 1) { System.err.println("Usage: <file>") System.exit(1) } val conf = new SparkConf() val sc = new SparkContext(conf) //SparkContext 是把代码提交到集群或者本地的通道,我们编写Spark代码,无论是要本地运行还是集群运行都必须有SparkContext的实例 val line = sc.textFile(args(0)) //把读取的内容保存给line变量,其实line是一个MappedRDD,Spark的所有操作都是基于RDD的 line.flatMap(_.split(" ")).map((_, 1)).reduceByKey(_+_).collect.foreach(println) sc.stop } }
b.导入spark包
File
->Project structure
->project settting
->libraries->+
导入spark-assembly-1.6.0-hadoop2.6.0.jar包(该包从spark安装包的lib下获得)
c.选择Artifacts
File
->Project structure
->project settting
->Artifacts->+,选择要导入的项目,以及main类
并且指定jar包输出的位置
d.输出jar包
Build -> Build ArtiFacts ->build,打好jar包到:D:\spark02\out\artifacts\spark02_jar\spark02.jar
e.上传jar包到spark客户端,并执行
执行命令:
spark-submit --master yarn --executor-memory 1000M /usr/local/tdr_hadoop/spark/spark02.jarhdfs://tdrHadoop/tmp/test/core-site.xml
在yarn的前台显示正在执行
执行结果输出:
本文为原创文章,同时在微信公众号发布,在转载请联系作者,想了解更多精彩大数据技术文章,请关注微信公众号:大数据梅峰谷 ,也可以扫描二维码
相关文章推荐
- 【spark 深入学习 05】RDD编程之旅基础篇-01
- 【Spark你妈喊你回家吃饭-01】 Spark是什么鬼?
- Spark快速大数据分析-RDD编程
- Spark学习笔记(一)--RDD编程
- Spark修炼之道(基础篇)——Linux大数据开发基础:第十四节:Shell编程入门(六)
- Spark修炼之道(基础篇)——Linux大数据开发基础:第十四节:Shell编程入门(六)
- Spark API编程动手实战-01-以本地模式进行Spark API实战map、filter和collect
- Spark RDD编程(Python和Scala版本)----Spark中的RDD就是一个不可变的分布式对象集合,是一种具有兼容性的基于内存的集群计算抽象方法,Spark则是这个方法的抽象。 Spa
- Spark RDD编程(Python和Scala版本)
- Spark RDD 编程实例
- Spark RDD/Core 编程 API入门系列之map、filter、textFile、cache、对Job输出结果进行升和降序、union、groupByKey、join、reduce、lookup(一)
- spark架构设计&编程模型01
- Spark RDD/Core 编程 API入门系列之动手实战和调试Spark文件操作、动手实战操作搜狗日志文件、搜狗日志文件深入实战(二)
- Spark RDD/Core 编程 API入门系列 之rdd实战(rdd基本操作实战及transformation和action流程图)(源码)(三)
- Spark的RDD编程(二)
- Spark API编程动手实战-01-以本地模式进行Spark API实战map、filter和collect
- spark内核揭秘-09-RDD的count操作 触发Job全生命周期-01
- Spark学习之RDD编程(2)
- Spark菜鸟学习营Day3 RDD编程进阶