spark相关知识点

spark相关知识点

1.spark是一站式的数据分析引擎

spark是一个计算组件，只做计算，不能做存储。里面有一个独立的资源管理系统（可以代替yarn，但一般没人用）

sparkRDD：编程的方式做数据分析，取代mapreduce

sparksql：取代hivesql（两者语句相同）

sparkStreaming：取代storm 做实时流

sparkml取代机器学习

2.四大特性

（1）快速：（spark比Hadoop快100倍，实际10倍左右）（因为spark在内存，hadoop在磁盘）

（2）易用：（可以支持多语言进行开发）

（3）通用性：（实时、离线、sql、机器学习）

（4）跨平台：（分配资源、各个地方读数据源）可以运行在：hadoop（yarn）、Apache Mesos、standalone（spark）。数据源多样性：scala集合、mysql、hive、habse、可以从任何能看到的数据系统都能做数据源

计算引擎本质：有向无环图

3.RDD 弹性分布式数据集【spark 数据结构】

弹性：数据既可以在内存也可以在磁盘（需要配置）（spark是尽量基于内存的计算框架）

数据集：数据集的来源有scala集合、hdfs文件、mysql，它们都可以转变成RDD。

sparkRDD组件能从外部加载数据源转换为spark能够识别的数据模型（围绕这个数据模型进行数据分析）

分布式：转换RDD后是分布式存储的，存储单元是partition

默认一个hdfs的block块就是一个RDD分区，所有的分区组合在一起就是一个RDD，就是一个完整的数据集。

补充：sparkRDD编程只能用高阶函数，所以不是面向对象是面向函数编程。

4.sparkRDD的操作

sparkRDD支持两种类型的操作：Transformation类型|Action类型

补充：spark每一个高阶函数都叫做算子

（1）Transformation类型：

输入算子（textFile：读取文本、parallelize：把scala集合变成RDD）

转换算子（map、flatmap、reduceByKey）

缓存算子（cache【内存】、persist【磁盘、内存、序列化或者非序列化、RDD缓存的时候做备份】）

注：transformation类型的算子分为shuffle型(触发stage划分)和非shuffle型。shuffle型算子一般需要让RDD的数据经过网络传输，非shuffle型算子在使用时一般不需要跨网络传输数据，所以尽量使用非shuffle型算子。

（2）Action类型：

一般输出型的算法（打印、持久化磁盘）

foreach()/count()/collect()/reduce()/saveAsHadoopFile()

一般是拉取形成一些集合

一般情况下调用了一个算子，如果返回值类型是RDD一般是transformation算子、如果不是RDD类型，一般为Action算子。

val map = List(1,2,3).map(t=>t+"_1") println(map)//scala中会执行，spark中不会执行

只有遇到action类型才会触发程序执行，不然不会执行（懒加载）【stage划分<->task】

spark程序以stage作为最小单元执行.

5.案例实现

案例一：spark版本的wordcount

public static void main(String[] args){ //1.spark任何一个组件 程序的入口 SparkConf conf = new SparkConf(); //配置文件信息 conf.setMaster("local[1]"); //本地执行 conf.setAppName("wordcount"); JavaSparkContext jsc = new JavaSparkContext(conf); //2.初始化RDD JavaRDD<String> linesRDD = jsc.textFile("C:\\Users\\Administrator\\Desktop\\javaclass\\spark\\ww.txt"); //3.各种数据分析 JavaRDD<String> flatMapRDD = linesRDD.flatMap(new FlatMapFunction<String, String>() { public Iterator<String> call(String line) throws Exception { return Arrays.asList(line.split(",")).iterator(); } }); JavaPairRDD<String, Integer> mapToPairRDD = flatMapRDD.mapToPair(new PairFunction<String, String, Integer>() { public Tuple2<String, Integer> call(String word) throws Exception { return new Tuple2<String, Integer>(word, 1); } }); JavaPairRDD<String, Integer> reduceByKeyRDD = mapToPairRDD.reduceByKey(new Function2<Integer, Integer, Integer>() { public Integer call(Integer v1, Integer v2) throws Exception { System.out.println(v1+" "+v2); return v1 + v2; } }); //4.结束 打印 持久化MySQLfs/hive/file reduceByKeyRDD.foreach(new VoidFunction<Tuple2<String, Integer>>() { public void call(Tuple2<String, Integer> tuple) throws Exception { System.out.println(tuple._1+" "+tuple._2); } }); jsc.stop(); }

案例二：jdk1.8版本的wordcount

public static void main(String[] args){ SparkConf conf = new SparkConf(); conf.setMaster("local[1]"); conf.setAppName("wordcount"); JavaSparkContext jsc = new JavaSparkContext(conf); JavaRDD<String> linesRDD = jsc.textFile("C:\\Users\\Administrator\\Desktop\\javaclass\\spark\\ww.txt"); //数据分析 JavaRDD<String> flatMapRDD = linesRDD.flatMap(line -> Arrays.asList(line.split(",")).iterator()); /** * 函数式编程 实现函数：输入-->输出 * 输入：上一个RDD javaRDD["i","jump"] * 输出 * * def mapToPair[K2, V2](f: PairFunction[T, K2, V2]): JavaPairRDD[K2, V2] = { * def cm: ClassTag[(K2, V2)] = implicitly[ClassTag[(K2, V2)]] * new JavaPairRDD(rdd.map[(K2, V2)](f)(cm))(fakeClassTag[K2], fakeClassTag[V2]) * } * @FunctionalInterface * public interface PairFunction<T, K, V> extends Serializable { * Tuple2<K, V> call(T t) throws Exception; * } */ JavaPairRDD<String, Integer> mapToPairRDD = flatMapRDD.mapToPair(word -> new Tuple2<String, Integer>(word, 1)); /** * @FunctionalInterface * public interface Function2<T1, T2, R> extends Serializable { * R call(T1 v1, T2 v2) throws Exception; * } */ JavaPairRDD<String, Integer> reduceByKeyRDD = mapToPairRDD.reduceByKey((v1, v2) -> v1 + v2); reduceByKeyRDD.foreach(t-> System.out.println(t._1+" "+t._2)); jsc.stop(); }

案例三：scala版本

def main(args: Array[String]): Unit = { val conf: SparkConf = new SparkConf() .setMaster("local") .setAppName("scala_wordcount") val sc: SparkContext = new SparkContext(conf) //scala 列表 val list = List( "i,jump", "you,jump" ) val listRDD: RDD[String] = sc.parallelize(list)//集合变为RDD /** * 接下来可以使用各种spark高阶算子做数据分析了 */ listRDD.flatMap(t=>t.split(",")) .map(word=>(word,1)) .reduceByKey((num1,num2)=>num1+num2) .foreach(println) //简化代码 //listRDD.flatMap(_.split(",")).map((_,1)).reduceByKey(_+_).foreach(println) sc.stop() }

6.宽依赖和窄依赖

宽依赖：RDD中partition中的数据，如果依赖父RDD多个分区（跨节点）【bykey】，叫宽依赖

窄依赖：RDD中partition中的数据，如果只依赖父RDD的一个分区、常量级的分区，就叫窄依赖。

常量级的分区一般出现在分区的合并（一般不跨节点）

sc.textfile.flatmap.map.reduceByKey.foreach():血缘关系链

从血缘关系链中可以看出stage划分

注意：

var rdd1 = sc.textfile.flatmap var rdd2 = rdd1.map //rdd1只会被消费一次 var rdd3 = rdd1.map //会从sc.textfile.flatmap开始再执行一遍而不是从rdd1开始，除非做了持久化

.map.reduceByKey.foreach():血缘关系链

从血缘关系链中可以看出stage划分

注意：

var rdd1 = sc.textfile.flatmap var rdd2 = rdd1.map //rdd1只会被消费一次 var rdd3 = rdd1.map //会从sc.textfile.flatmap开始再执行一遍而不是从rdd1开始，除非做了持久化