spark-01_spark中间结果

1.   spark优势

1）  和hadoop比较

MR问题:

    a.频繁落地，每一步中间结果都落地（基于硬件和网络都很差环境，设计时很注重数据安全传递）-按现在硬件和网络情况“过渡设计”map

    b.大的计算是有很多步骤，如果每一步都落地，计算不能迭代运行

文件太大，1g，海量数据1TB，

海量

 

spark解决

1）  mr频繁落地，落地还是需要，但是可以部分过程不落地，也就是说，落地次数减少

2）  这个内存中连续计算部分，就可以迭代运行

spark运行速度是hadoop100倍。

 

spark和hadoop不是替代关系

hadoop生态链中，核心有三个内容：MR（hive，shark,spark-sql）离线计算，HDFS分布式存储，YARN资源调度

spark并发计算在线、实时，基于HDFS，基于YARN（几千节点）、ApacheMESOS（3w+）

总结：相辅相成关系，它会替代掉MR，hive，spark-sql，实时分析

 

2）  hadoop和spark

hadoop稳定。

spark新技术：RDD

2. Spark提供几种集群模式：

1）  Standalone自身

2）  Yarn整合

3）  Mesos整合

3.Spark和Storm实时计算框架

两个最重要指标：1）实时2）吞吐量（批量）

实时性：storm快，亚秒级0.x秒，spark，1秒

吞吐量：storm事件，一条一条处理；spark时间，堆积一部分后进行处理（批量）。

目前应用storm比spark多。

 

R语言，科技计算，写好了很多类库。

4.RDD

1）弹性分布式数据集，本质是函数

function add( int a, int b){

         returna+b;

}

add(a,b)

一条一条读数据

1）  RDD组成

总共有4部分，核心有3部分，

数据（引用），计算函数（函数式编程），血缘（lineage）

5.RDD操作分两类：

1）  键值对pair，map/tuple

2）  非键值对

 

6.RDD操作分两类：

1）  transformations

2）  actions

a)   是否发生shuffle根据上面的类型来判断？

完全靠上面分类来说是否发生shuffle，这是不对的；

大多数的transformations类型的命令不发生shuffle，大多数actions类型的命令会发生shuffle。

 

b)  迭代：

transformations命令迭代执行，它就可以应用spark流水线思想，并发执行，执行过程不落地。上一步执行结果，可以为下一步提供数据，继续运行。

actions，一碰到actions命令，spark就自动将命令划分成多个stage，在一个stage内的命令可以批量执行。迭代计算。跨域stage，前一个stage会发生shuffle写，后一个stage会发生shuffle读，但不是所有情况。

7.在一个stage内有多个命令，命令不会立即执行

懒加载，懒执行 lazy

8.宽窄依赖，它才是决定是否发生shuffle的依据

1）  窄依赖不会发生shuffle

2）  宽依赖必然发生shuffle

 

一个RDD来自一个父RDD（窄依赖，不会发生shuffle）（一对一）

一个RDD的内容来自两个父RDD（窄依赖，不会发生shuffle）（多对一）

执行命令前就决定了shuffle是否发生。

上面的情况不会shuffle（多对一）

上面情况，多对多的依赖，这时才发生shuffle

一个RDD的数据不能来自多个父的RDD，数据交叉，发生shuffle

9.map和join算子

a)   map会不会发生shuffle

 

b)  join操作，发生shuffle

 

shuffle是在全部命令提交后，spark进行分析后，已经决定（命令自身就决定是否发生shuffle）

shuffle就会落地，就影响性能，实际应用中要避免shuffle。不能完全避免

10.DAG有向无环图

目的：spark自动分配任务，自动产生，根据命令定义

1）stage划分出来的？跟命令类型，根据是否发生shuffle。如果遇到shuffle命令，就会把前面的命令放在一个stage内，把后面继续划分。

2）stage内的命令就可以应用spark流水线思想，迭代批量的执行

3）每个新的RDD会自动划分到一个stage内

 

avro spark中默认序列化

11.cache和persist算子

cache,persist 非常特殊，不属于transformations或者actions命令。

RDD只读，如果在一个stage内大量transformations类型命令，可能造成内存耗尽情况，因为每步中间结果都需要产生新的RDD，然后还一直放在内存中。万一内存不够，可以人为加入shuffle，中间结果的数据落地。性能有一定损失。

cache和persist命令差异，persist多了一个压缩。

12.scala语言

1）  spark是用scala语言写的，neo4j图形用scala，Kafka消息队列，storm

 

市场主流一些并行开发的语言

go/r/erlang/scala

13.java和scala的关系

java是c的儿子；scala是java的儿子

*.java –> *.class -> JVM

*.scala ->*.class –> JVM

 

scala可以使用自己的资源，还可以无缝使用java所有的api

Java的JVM的作者之一就是scala作者

14.开发环境工具

1）  windows scala （scala-2.11.7.msi）安装不要安装在中文目录，不要有空格

2）  windows/linux  eclipse forscala（scala-SDK-4.2.0-vfinal-2.11-win32.win32.x86_64.zip）：d2pd

3) .CLASS文件查看工具

15.scala基础

java：public void saveItem(Item item){ }

scala：def saveItem(item:Item): Unit = { }

 

java：public List<Item> findItemList(Itemitem){ }

scala：def findItemList(item:Item):List[Item] = {}

（1）创建一个Scala Project工程

（2）修改jdk

（3）创建一个Scala Object（特点：单例）

（4）编码

//object 天生单例，不需要像java一样需要懒汉式、饿汉式
object Hello {
  //定义一个方法，方法名是main，入口方法
  //识别定义方法：def开头，有等号
  def main(args: Array[String]): Unit = {
    println("zuo li liang")
  }
}

源码：

16.java变量使用多， scala 常量使用多

常量一次赋值，赋值后不能在变化；

变量可以多次赋值，但赋值之后之前旧的值就不存在；

scala采用常量，计算完的结果设置到新的常量中。

spark中每一步加工后会产生一个新的RDD。RDD只读

 

常量，不允许修改，这样后面计算要用到前面计算的中间结果，直接获取就行了。

新的值就产生一个新的常量来存放。

 

解决，共享变量的线程安全问题。

session共享，链接是有状态（登录）和无状态。

java编程算法在本地；scala数据和算法一起传递，把有状态的请求就变成无状态。（函数式编程）

17.var变量，val常量

java变量，8种基本变量，引用；scala中不区分变量类型，都定义成var/val。它会自动类型推导。

 

java的string，

String s = “hello”;

s += “spark”;

产生3个字符串，它不是简单变量，对象，引用。

 

在scala中定义s，StringContext

18.lazy懒执行，懒加载

1）lazy后常量，没有执行，没有分配空间。

2）transformations类型的命令都是这样做的，RDD

 

函数

没有def，用=>分割，它的前面，参数列表，它的后面是一个函数体

(n:Int) => {n%2!=0} 这个没有名称，所有称为匿名函数，自己调用别人无法调用（因为没有名称，想调用也无法调到）

 

scala中使用{}或者()都可以，如果括号太多，最好使用{}

 

匿名函数有什么好处？

非常灵活，随想随用。

19.Option

有两个返回值Some(n) n就代表执行结果（执行成功）

如果执行出错，None

20.Array和List不同？

Array数组，内存地址连续，读取速度快

List链表，内存地址不连续，插入元素快

Array和List都是常量；ArrayBuffer和ListBuffer变量

21. Java类和Scala类的不同

Java：

Class/Interface/Abstract

          Class：没有类java就启动不了

         Interface：(1)让类不直接暴露给用户，类的方法中虽然都是公用的，当有些方法不想让用户a知道，有些方法不想让用户b知道，接口可以实现这个屏蔽功能。（2）接口形成规范后换实现类比较容易

         Abstract：有些方法是公用的有些方法是你自己去实现，最典型的就是spring

它将每个类中公用的方法都写好，一少部分需要用户自己去实现，这样一结合就可以完成一个功能。

Scala:

Object单例/Class类/Case Class样例类/Trait 抽象类

（1）Scala Class

特性：和java的结构相同，和java最大的不同是属性的类型不同，一般是val（可以是var但是不推荐使用，因为将来分布式的时候都有共享变量问题，如果是var，到时候需要好多方面来协调这个信息，很容易出错，而val是只读，就不会有共享变量问题，也就不会有线程安全问题）

创建工程

编码：加val

源码：

就是java代码：没有set方法，其中的name、age方法相当于get方法

调用：

修改编写的代码：不加val

源码：

调用：

修改编码：加var

源码：

调用：

（2）CaseClass样例类

特性：天然序列化，在实际运用中用到的比较多，因为大部分写的程序需要通过网络传输，就势必要序列化，而caseclass天然就序列化很符合

package com.liming
case class Teacher(var name:String,age:Integer){
  def teach()={
    println(s"老师:${name}正在上课。。。。。。")
  }
  def add(age:Int)={
    println("add"+age)
  }
}
 
object Teacher {
  def main(args: Array[String]): Unit = {
    val t = Teacher("Taylor Swift",18)//创建对象省略new关键字
    println(t.name)
    t.name = "霉霉"
    println(t.name)
    t.teach()
    
    val t2 = Teacher("谢春花",30)
    t2 add 1 //当方法是对象的属性时，括号可以去掉
  }
}

（3）extend继承

package com.liming
class Person(val name:String){
  
}
class Student1(name:String,val age:Integer) extends Person("Taylor Swift"){
  
}
class Student2(override val name:String,val age:Integer) extends Person(name){
  
}
object Student {
  def main(args: Array[String]): Unit = {
		  val s1 = new Student1("a",18)
		  println(s1.name)
		  val s2 = new Student2("b",17)
		  println(s2.name)
  }
}

（3）Trait接口、抽象类

package com.liming
 
 
trait Work {
  def todo(); //抽象方法，待子类实现
  def doing()={
    println("doing.....")
  }
}
case class task() extends Work{
  def todo()={
    println("to do something")
  }
}
object test{
  def main(args: Array[String]): Unit = {
    //val o = work  //抽象类不能实例化
    val t = new task()
    t.doing()
    t.todo()
  }
}