新建一个 Scala 类
Transformation 从一个 RDD 构建新的 RDD
RDDs 介绍: 弹性分布式数据集,所有的计算都是通过 RDDs “创建-转换-完成”的
.textFile("/xx" ) 导入的文件过大时,会自动分割文件,在集群中分割
------------------------------------------------
.parallelize(x,x) 方法,将已存在的集合对象,按分区分片导入 RDD 对象
Scala 脚本语法: val 常量 var 变量 (可重新指向,但必须是同类型)
------------------------------------------------
Scala 语法: 匿名函数当做参数 .filter=>(line.contains("xx"))
演示了 ssh 不需要输密码的修改
执行 ssh-keygen (一路回车)
执行 cd .ssh 看到几个文件,其中 .pub 为公钥
创建文件 authorized_keys 文件
执行 cat 文件 > 文件 (把那个 .pub 导入 authorized_keys )
chmod 600 authorized_keys
验收:再执行 ssh localhost 看到启动后不需要输入密码
---------------------------------------------------------------
给集群提交作业,也是把作业(工程)打包成 jar 然后上传到服务路径
./bin/spark-submit .... xx.jar (工程)
scala 在 idea 里新建 java 项目,在 idea 搜素插件 scala 并安装(已经安装了)
file -> new project -> sbt (表示用 sbt 打包) 等待生成各种目录和文件
版本匹配: scala , spark , jdk , sbt 的版本都要匹配
spark 基于 scala 运行在 jvm 环境中 1.7版本以上
spark 不依赖 hadoop
目录:
bin 可执行文件
core streaming python 包含组件源代码
examples 包含单机 job 例子,以供研究
执行命令:
./spark-shell (等待时间较长)
在执行成功后可看到 scala> 提示符,然后就可以使用 python 进行交互了
hadoop 缺点:离线处理,导致时效性差,几分钟几小时为单位
spark 有点:时效性高,在内存里进行
Spark Core :任务调度,内存管理,容错机制
RDDs : 弹性分布式数据集
Spark SQL:处理结构式数据
Mlib: 机器学习相关的包(支持集群扩展,python 缺点是单机性能有限)
Spark 快速且通用的集群计算平台
-快速,比hadoop快,基于内存的 MapReduce(计算结果中间数据在内存中)
-通用,应用场景更多,批处理,迭代计算,交互查询,流处理
由 scala 语言编写,但是 python java 也支持
combineByKey():
(createCombiner, mergeValue, mergeCombiners, partitioner)
最常用的基于key的聚合函数,返回的类型可以与输入类型不一样。
许多基于key的聚合函数都用到了它,像groupByKey()
遍历partition中的元素,元素的key,要么之前见过的,要么不是。
如果是新元素,使用我们提供的createCombiner()函数
如果是这个partition中已经存在的key,就会使用mergeValue()函数
合并每个partition的结果的时候,使用mergeCombiners()函数
keyvalue对rdds:
创建keyvalue对rdds,使用map()函数,返回key/value对
例如,包含数行数据的rdd,把每行数据的第一个单词作为keys。
RDDs血统关系图
Spark维护着RDDs之间的依赖关系和创建关系,叫做 血统关系图
Spark使用血统关系图计算每个RDD的需求和恢复丢失的数据
dds的特性
延迟计算:
spark对rdds的计算是,他们第一次使用action操作的时候;
这种方式在处理大数据的时候特别有用,可以减少数据的传输;
spark内部记录metadata 表名tranformations操作已经被响应了;
加载数据也是延迟计算,数据只有在必要的时候,才会被加载进去。
rdd基本操作之action介绍,
action表示在rdd上计算出来一个结果。
把结果返回给driver program或保存在文件系统,count()、save都算是基本的action。
集合运算
RDDs支持数学集合的计算,例如并集、交集等
rdd transformations(转换二)操作:集合运算
1.distinct :去除重复
2.subturast: rdd1.subtruast(rdd2),rdd1中有的,而rdd2中没有的
3.union。并集:rdd1.union(rdd2),取rdd1和rdd2中所有的数据
4.innter:交集:取rdd1与rdd2相同的部分
val rdd_distinct = rdd1.distinct()
val rdd_union = rdd1.union(rdd2)
val rdd_inter = rdd1.intersection(rdd2)
val rdd_sub = rdd1.subtract(rdd2)
Map() Filter() Flatmap()压扁
rdd transformations(转换二)操作:集合运算
1.distinct :去除重复
2.subturast: rdd1.subtruast(rdd2),rdd1中有的,而rdd2中没有的
3.union。并集:rdd1.union(rdd2),取rdd1和rdd2中所有的数据
4.innter:交集:取rdd1与rdd2相同的部分
5. subtract 包含 rdd1有rdd2的没有的部分
Transformations(转换)
从之前的RDD构建一个新的RDD,像map()和filter()
逐元素Transformations:
map()接收函数,把函数应用到RDD的每一个元素,返回新RDD
filter()接收函数,返回只包含满足filter()函数的元素的新RDD
flatMap(),对每个输入元素,输出多个输出元素。flat压扁的意思,将RDD中元素压扁后返回一个新的RDD
val lines = inputs.flatMap(line=>line.split(" "))
lines.foreach(println)
Scala的匿名函数
Scala的匿名函数
Driver program:包含了main方法
sparkContext:链接spark集群,一般直接使用shell创建好的对象sc即可
rdds:弹性分布数据集,spark如果大数据量时会自动将片分配到每一个节点上,各个片的总名称就是rdds 对象
创建rdds的两种对象:
1.使用sc.prilize对象
2.使用sc.textFile()加载外部对象
3.scala:匿名函数和类型推断。
line2.Filter(line=>line.contains("world"))
即line2通过Filter的方法将每一行赋值给line,然后line在调用contains方法确认哪一行是否包含有world,这个单词,有则返回给line
RDD不可改变的分布式数据集合对象
所有的计算都是RDD的生成,转换操作完成
一个RDD内部由许多partitions分片组成
分片是并行处理单元, spark底层会并行的,或者顺行的处理。
RDDs创建#scala
val rdd= sc.parallelize(Array(1,2,3,4),4)
#参数1:待并行化处理的集合
#参数2:分区个数
rdd.count() //计算个数
rdd.foreach(println) //遍历
#加载外部数据
val rddText = sc.textFile("hello.txt")
Scala基础#变量
val 变量值不可修改
var 可指向类型相同的值
#匿名函数和类型推断 自动推断line是string类型,可调用contain方法
lines.filter(line => line.contains("world"))
spark的运行环境:
基于Scala ,运行在JVM,运行环境Java7+
spark下载:
搭建spark不需要Hadoop,下载后解压
虚拟机(Linux)联网状态下,通过 wget+下载链接
Linux解压命令 tar -zxvf spark.....
spark目录
bin目录-可执行文件
core,streaming主要包含组件源代码
examples 包含单机例子
spark的shell:
梳理分布在集群上的数据
可以将数据加载到节点的内存中,因此分布式处理可在秒级完成。
快速迭代计算,实时查询、分析
spark提供了Python shell和Scala shell
Scala shell
/bin/spark-shell
Hadoop应用场景:离线处理、对时效性要求不高(执行时间几分钟到几小时不等);
Spark应用场景:时效性要求高的场景、机器学习等领域。
两者比较:
这是生态系统,每个组件都有其作用,各善其职即可;
Spark不具有HDFS的存储能力,要借助HDFS等持久化数据;
大数据将会孕育出更多的新技术
spark core:
包含spark的基本功能,任务调度,内存管理,容错机制
内部定义了RDDs,弹性分布式数据集
spark sql:
是spark处理结构化数据的库
spark streaming:
实时数据流处理组件
milb:
包含通用机器学习功能的包,分类,聚类,回归
支持起群上的横向扩展
graphx:
处理图的库,并进行图的并行计算
cluster managers:
集群管理
紧密集成的优点
spark底层优化了,基于spark的底层组件也会得到相应的优化。
紧密继承节省了各个组价组合使用时的部署,测试等时间
向spark增加新的组件时,其他组件,可立刻享用新组建的功能。
Spark快速且通用的集群计算平台
➢ spark是快速的
○ Spark扩充了流行的Mapreduce计算模型
○ Spark是基于内存的计算
➢ spark是通用的
○ 容纳了其他分布式系统拥有的功能:批处理、迭代式计算、交互查询和流处理等,方便维护
○ 优点:降低了维护成本
➢ spark是高度开放的
○ Spark提供了Python,Java,Scala,SQL的API和丰富的内置库。
○ Spark和其他的大数据工具整合的很好,包括hadoop,kafka等。
RDDs血统关系图
Spark维护着RDDs之间的依赖关系和创建关系,叫做 血统关系图
Spark使用血统关系图计算每个RDD的需求和恢复丢失的数据
延迟计算( Lazy Evaluation)
在第一次使用Action操作时才进行计算, 减少数据传输
Spark内部记录metadat表明 transformation操作已经相应
RDD.persist() 持久操作
默认每次RDD进行action操作,会重新计算
persist()后可以重复利用一个RDD (缓存)
//scala //RDD逐元素transformation lines.map(word=>(word,1)) lines.filter(word=>word.contains("hello")) lines.flatMap(line=>line.split(" ")) //压扁 //集合运算 rdd1.distinct() //去重 rdd1.union(rdd2) //并集 rdd1.intersection(rdd2) //交集 rdd1.subtract(rdd2)
RDDs创建
#scala val rdd= sc.parallelize(Array(1,2,3,4),4) #参数1:待并行化处理的集合 #参数2:分区个数 rdd.count() rdd.foreach(println) #加载外部数据 val rddText = sc.textFile("hello.txt")
Scala基础
#变量 val 变量值不可修改 var 可指向类型相同的值 #匿名函数和类型推断 自动推断line是string类型,可调用contain方法 lines.filter(line => line.contains("world"))
Drive Programs通过 SparkContext 对象访问Spark
SparkContext 对象(即sc) 代表和一个集群的连接
scala> val lines= sc.textFile("/home/soft/hello.txt") lines即为RDDs
RDDs弹性分布式数据集: 并行分布在整个集群中
RDDs是Spark分发数据和计算的基础抽象类
一个RDD是不可改变的分布式集合对象
Spark中所有计算都是RDD操作完成
分片:
一个RDD内部有许多partitions分片组成,
每个partition包含一部分数据, 可在集群不同节点计算
分片是Spark并行处理的单元
Wordcount
//Scala Object WordCount{ def main(args: Array[String]){ val conf= new SparkConf().setAppName("wordcount") val sc = new SparkContext(conf) val input= sc.textFile("/home/soft/hello.txt") //RDD操作:压扁 val lines = input.flatMap(line=> line.split(" ")) //转换成kv对 val count= lines.map(word=>(word,1)).reduceByKey{case (x,y)=>x+y} val output= count.saveAsTextFile("/home/result") } }
Project Structure -> Artifacts ->+然后 BuildArtifacts 打包Jar
启动集群:
启动master start-master.sh
启动worker spark-class
提交作业 spark-submit
#启动worker spark-class org.apache.spark.deploy.worker.Worker spark://localhost.localdomain:4040 #提交 spark-submit --master spark://localhost.localdomain:4040 --class WordCount /home/soft/hello.jar #上传jar包 rz -be
RDDS的特性
RDD基本操作之action
rdd缓存级别