新建一个 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缓存级别