Spark简述

Spark 简介

Spark是什么
一个快速且通用的集群计算平台

Spark特点：
 1. Spark扩充了流行的Mapreduce计算模型
 
 2. Spark是基于内存的计算（一些中间数据存储在内存中，加快读写）
 
 
Spark 是通用的
    Spark的设计容纳了其他分布式系统拥有的功能
    
    批处理，迭代式计算，交互查询和流处理等
    
Spark是高度开放的

    Spark提供了Python Java Scala Sql的API和丰富的内置库
    Spark和其他的大数据工具整合的很好，包括Hadoop、kafaka等

1. Spark生态圈

最初是基于Hadoop MapReduce的
发现MapReduce在迭代式计算和交互式上低效，引入了内存存储（重大的区别）

Spark的组件：
Spark SQL structured data：
是Spark处理结构化数据的库，就像Hive SQL，Mysql一样--报表统计
Spark Streaming real-time：
是实时数据流处理组件，类似Storm
Spark Streaming提供了API来操作实时流数据
应用场景，企业中用来从Kafka接收数据做实时统计




Mlib machine learning

一个包含通用机器学习功能的包， Machine learning lib
包含分类，聚类，回归等，还包括模型评估，和数据导入

MLib提供的上面这些方法，都支持集群上的横向扩展





Graphx graph processing

是处理图的库（例如，社交网络），并进行图的并行计算
继承了RDD API
它提供了各种图的操作和常用的图算法，例如PangeRank算法
应用场景，图计算

---------------------
Spark Core
包含了Spark的基本功能，包含任务调度，内存管理，容错机制
内部定义了RDDs（弹性分布式数据集）
提供了很多APPIs来创建和操作这些RDDs
应用场景，为其他组件提供底层服务

---------------------
Cluster Managers：
    
    就是集群管理，Spark自带一个集群管理是单独调度器
    常见集群管理包括Hadoop YARN， Apache Mesos

紧密集成的优点
Spark底层优化了，基于Spark底层的组件，也得到了相应的优化

紧密集成，节省了各个组件组合使用时的部署，测试等时间

向Spark增加新的组件，其它组件时，可立刻享用新组件的功能

2. Spark 和 Hadoop比较

Hadoop应用场景：
离线处理

对时效性要求不高：中间数据会落在硬盘上（几分钟到几小时不等）

Spark应用场景：
时效性要求高的场景（几秒钟到几分钟不等）

机器学习等领域
基于内存

比较：
这是生态系统，每个组件都有其作用，各司其职即可

Spark不具有HDFS的存储能力，要借助HDFS等持久化数据。
大数据将会孕育出更多的新技术

Spark安装
数据处理？？？？

Spark运行环境：
Spark是Scala写的，运行在JVM上，所有运行环境是Java7+
python 2.6+ 3.7+


Spark的Shell：
Spark的Shell使你能够处理分布在集群上的数据

Spark把数据加载到节点的内存中，因此分布式处理可在秒级完成

快速迭代式计算，实时查询，分析一般能够在shells中完成

Spark提供了Python shells 和 Scala shells

python shell ：
 bin/pyspark
 bin/spark-shell
 
 spark是用Scala写的？？？
 网络问题 版本问题

Rdds

RDDs介绍

Driver programe
 包含程序main()方法，RDDs定义和操作
 它管理很多节点，我们称作executors
 
SparkContext：
 driver programs 通过SparkContext对象访问Spark
 SparkContext对象代表和一个集群的连接
 在shell中SparkContext自动创建好了，就是sc变量（直接调用即可）
 scala> val lines=sc.textFile（../xx）
-----------------------------

RDDs：
1. Resilient distributed datasets（弹性分布式数据集，简写RDDs）
2. 这些RDDs，并行的分布在整个集群中
eg。1
scala> val lines=sc.textFile（../xx）
假设这个文件有500g ，对其切分分布在5个节点上
通过lines就可以知道文件存放在哪，并操作（lines就是RDDs）
3. RDDs是Spark分发数据和计算的基础抽象类

    1.1 一个RDD是一个不可改变的分布式集合对象（相当于文件被占用，不可修改）
    1.2 Spark中，所有的计算都是通过RDDs的创建，转换，操作完成的。
    1.3 一个RDD内部由许多partitions（分片）组成
    
分片：
每个分片包括一部分数据，partitions可在集群不同节点上计算

分片是Spark并行处理的单元，Spark顺序的，并行的处理分片

RDDs的创建方法：
一、 把一个存在的集合传给SparkContext的parallelize()方法，测试用
val rdd = sc.parallelize（Array（1，2，2，4），4）
第一个参数：待并行化处理的集合，第2个参数：分区个数

二、 加载外部数据集

val rddText = sc.textFile（“xxx.txt”）


Scala基础语法：
val  var创建变量
val：不可修改
var：可变

Scala的匿名函数和类型推断

RDD基本操作之Transformation

Transformation介绍：

从之前RDD构建一个新的RDD，像map（）和filter（）

逐元素Transformation
 map（）
 
 map接收函数，把函数应用到RDD的每一个元素，返回新RDD
 
 val lines = lines.map（word=>（word，1））
 
 filter() 
 
 filter（）接收函数，返回只包含满足filter（）函数的元素的新rdd
 
 
 
 val lines =lines.filter（word=>word.contains(‘hello’)）
 
 flatMap()
 
 
 val inputs = sc.TextFile("xxx.txt")
 
 val lines = inputs.flatMap(line=>line.split(" "))
 
 
集合运算
并集 交集等

val rdd1=sc.parallelize（Array（"coffe","coffe","panda","monkey","tea"））

val rdd2=sc.parallelize（Array（"coffe","coffe","panda","monkey","cat"））

val rdd_distinct=rdd1.distinct(rdd2)

val rdd_union =rdd1.union()

val rdd_inter=rdd.intersection(rdd2)

val rdd_sub=rdd1.subtract(rdd2)

RDD基本操作action，对返回结果对操作？？？
Action介绍
在RDD上计算出来一个结果
把结果返回给driver program或保存在文件系统，count（），save

reduce() //累加，计数，和其他类型的聚集操作
rdd.reduce((x,y)=>x+y)

Collect() 
//遍历整个RDD，向driver program返回RDD内容
//需要单机内存能够容纳下（因为数据要拷贝给driver，测试使用）
大数据的时候，使用saveAsTextFile() action等

take（n）
返回RDD的n个元素（同时尝试访问最少的partitions）
返回结果是无序的测试使用

top（）
//排序

foreach（）
//计算RDD中的每个元素，但不返回本地，不保存


RDDs的特性
RDDs的血统关系图：
 Spark维护着RDDs之间的依赖关系和创建关系，叫做血统关系图
 
 Spark使用血统关系图来计算每个RDD的需求和恢复丢失的数据
 
延迟计算（lazy evaluation）

Spark对RDDs的计算是，他们第一次使用action操作的时候
这种方式在处理大数据的时候特别有用，可以减少数据的传输
Spark内部记录metadata表明transformation操作已被响应
加载数据也是延迟计算，数据只有在必要的时候，才会被加载进去

RDD.persist()
 默认每次在RDDs上进行action操作时，Spark都重新计算RDDs
 如果想重复利用一个RDD，可以使用RDD.persist（）参数：内存，硬盘
 unpersist 从缓存中移除

KeyValue对RDDs

创建KeyValue对Rdds：
使用Map函数 ，返回keyValue

eg.1 
使用map（）函数，返回key/vlaule
例如，包含数行数据对RDD，把每行数据数据的第一个单词作为keys


val rdd2=rdd.map（line=>(line.split(" ")(0),line)）

KeyValue对RDDs的Transformations（example：{（1，2），（3，4），（5，6）}）

reduceBykey()
groupByKey()
.............


combineByKey()：
（createCombiner,mergeValue,MergeCombiners,partitioner）
最常用基于key的聚合函数，返回类型可以与输入类型不一样
许多基于key的聚合函数都用到了它，像groupByKey（）

combineByKey():
遍历partition中的元素，元素的key，要么之前见过，要么不是
如果是新元素，使用我们提供的createCombiner()函数
如果是这个partition中已经存在key，就会使用mergeValue()函数
合计每个partition的结果的时候，使用mergeCombiner()函数