Spark学习笔记

Data

Source->Kafka->Spark Streaming->Parquet->Spark SQL(SparkSQL可以结合ML、GraphX等)->Parquet->其它各种Data Mining等

1.1 Spark集群的安装

Spark的运行是构建在hadoop集群之上（默认hadoop集群已经安装好了），在spark集群集群上必须要安装对应版本的scala

1.1.1 scala安装

Ø下载scala版本，解压scala

Ø配置环境变量/etc/profile,添加SCALA_HOME、修改PATH，添加上scala的path路径

Ø进入$SCALA_HOME/bin目录，执行./scala验证scala是否安装成功

Ø集群机器都需要安装scala

1.1.2 spark安装

在集群的所有机器上都必须要安装spark，首先安装master的spark程序

Ø先解压spark程序

Ø修改环境变量/etc/profile添加SPARK_HOME和修改spark PATH路径

Ø配置spark，进入conf目录下

nmv spark-env.sh.template spark-env.sh

其中：spark_master_ip：用于指定master

nvi slaves修改文件，把work节点都添加进去;

Ø至此，spark集群安装完毕

1.1.3启动集群校验

Ø先启动hadoop集群，jps查看进程

Ø再启动spark集群，在sbin目录下执行./start-all.shjps查看进程

ØUi访问，检查集群情况http://master:8080

Ø进入spark/bin目录下，启动spark-shell脚本

1.2 spark-shell的使用

在master机器上的$SPARK_HOME/bin目录下，运行./spark-shell程序启动shark-shell脚本；通过http://master:4040查看spark-shell运行情况

1.2.1 spark-shell操作hdfs文件实战

Ø将spark目录下的README.md文件上传到hdfs上的/test目录下，通过hdfs ui来进行查看slave:50070/explorer.html#/查看文件是否上传成功

Ø在spark-shell脚本程序下，执行sc（SparkContext实例），启动spark-sehll时，系统自动生成

scala> sc

res0: org.apache.spark.SparkContext = org.apache.spark.SparkContext@65859b44

# SparkContext是把代码提交到集群或者本地的通道，编写Spark代码，无论是要运行本地还是集群都必须要有SparkContext实例

ØSpark-shell读取hdfs文件的README.md文件

val file = sc.textFile(“hdfs://mapeng:8020/test/README.md”)

#这里把读取到的文本内容赋值给了变量file,（就是一个MappedRDD，在spark的代码中，一切都是基于RDD进行操作的）

Ø读取文本中包含有“spark”的行

scala> val sparks = file.filter(line

=> line.contains("spark"))

sparks: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[3] at

filter at :26

#此时生成了一个FilterRDD

Ø统计spark一共出现多少次

Sparks.count

1.2.2 spark-shell操作及详细说明

1.2.2.1并行化集合（parallelize）

Ø加载集合数据

val data = sc.parallelize(1 to 10)#加载集合数据

或者; val data = sc.parallelize(List(1,2,3,4…))

Ø对集合数据进行*2操作

val data1 = data.map(_*2)

Ø对数据进行过滤：过滤出是2的倍数的集合

val data2 = data.filter(_%2==0)

Ø内存缓存数据

data.cache

Ø触发action，以数据的形式返回结果集

data.collect

Ø返回结果集的第一个元素

data.first

Ø返回结果集的前3个元素

data.take(3)

Ø统计元素的个数

data.count

Ø查看RDD的转换过程

data.toDebugString

1.2.2.2 map数据集合

Ø加载List(Map)数据

val

data=sc.parallelize(List(("A",1),("B",2),("C",3),("A",4),("B",5)))

Ø排序sortByKey()

scala> data.sortByKey().collect

res55: Array[(String, Int)] = Array((A,1), (A,4), (B,2), (B,5),

(C,3))

Ø分组groupByKey()

scala>

data.groupByKey().collect

res57: Array[(String, Iterable[Int])] = Array((B,CompactBuffer(2,  5)), (A,CompactBuffer(1, 4)), (C,CompactBuffer(3)))

Ø求和reduceByKey(_+_)

scala>

data.reduceByKey(_+_).collect

res59: Array[(String, Int)] = Array((B,7), (A,5), (C,3))

Ø去重distinct

scala> data.distinct.collect

res60: Array[(String, Int)] = Array((A,1), (A,4), (B,5), (C,3),

(B,2))

Ø联合union

scala> val

data1=sc.parallelize(List(("A",1),("B",2),("C",3),("A",4),("B",5)))

data1:

org.apache.spark.rdd.RDD[(String, Int)] = ParallelCollectionRDD[0] at

parallelize at :24

scala> val data2=sc.parallelize(List(("A",4),("A",4),("C",3),("A",4),("B",5)))

data2:

org.apache.spark.rdd.RDD[(String, Int)] = ParallelCollectionRDD[1] at

parallelize at :24

scala>

data1.union(data2).collect

res1: Array[(String, Int)] = Array((A,1), (B,2), (C,3), (A,4),

(B,5), (A,4), (A,4), (C,3), (A,4), (B,5))

Ø关联join

相当于笛卡尔积

1.2.2.3保存转换结果saveAsTextFile

data.saveAsTextFile(“path”);//将转换结果存储在hdfs指定的路径

1.2.3 spark cache缓存

对于spark程序，第二次执行要比前面的执行的效率要高

1.3 RDD(弹性分布式数据集)

1.3.1 RDD介绍

ØRDD是一个容错的、并行的数据结构，可以让用户显示的将数据存储在磁盘和内存中，并能控制数据的分区。

ØRDD提供了一套丰富的函数来操作数据

ØRDD作为数据机构，本质上是一个只读的分区记录集合；一个RDD可以包含多个分区，每个分区就是一个dataSet片段；RDD可以相互依赖

n窄依赖：RDD的每个分区最多只能被一个child

RDD的分区使用（例如：map操作）

n宽依赖：RDD的分区可以被多个child

RDD的分区使用（例如：join操作）

区别：

（1）窄依赖可以在集群中的一个节点上如流水般的执行，相反，宽依赖需要所有的父分区的数据都可用

（2）从出现失败恢复的角度来考虑：窄依赖只需要重新计算失败的父RDD的分区，而宽依赖失败会导致其父RDD的多个分区重新计算

1.3.2 RDD分区

1.3.3创建操作

1.3.3.1集合创建操作

Spark提供了两类函数实现从集合生成RDD；

Øparallelize

val rdd = sc.parallelize(1 to 100)

ØmakeRDD:还提供了指定分区参数

val rdd = sc.makeRDD(1 to 100,3)#指定了分区数为3

1.3.3.2存储创建操作

操作hdfs

val rdd = sc.textFile(“hdfs://master:9000/test/xxx.txt”)

1.3.4 RDD的基本转换操作

1.3.4.1 RDD的重新分区

repartition和coalesce是对RDD的分区进行重新划分

Ørepartition(numPartitions:Int):RDD[T]

Øcoalesce(numPartitions:Int,shuffle:Boolean=false):RDD[T]

repartition只是coalesce接口中shuffle为true的简易实现。

重新划分分区主要有三种情况：（原RDD有N个分区，需要重新划分为M个分区）

ØN

ØN>M(相差不大)：面临着要把原分区进行合并的操作，最终合成M个分区，这时将shuffle设置为false

注：在shuffle为false时，设置M>N，coalesce是不起作用的

ØN>>M（差距悬殊）：如果将shuffle设置为false，由于父子RDD是窄依赖，会使得它们同处于一个stage中，可能会造成spark程序运行的并行度不够，从而影响效率。

因而，最好设置为true，使得coalesce之前的操作有更好的并行度

1.3.4.2 RDD转换为数组（randomSplit、glom）

ØrandomSplit(weight:Array[Double],seed:Long=System.nanoTime):Array[RDD[T]]

randomSplit函数是将一个RDD切分为多个RDD，返回结果是一个RDD数组;函数的第一个方法传入的参数权重是一个Double类型的数组；权重大的，分到的数据的概率大

val rdd = sc.makeRDD(1 to 10)

val splitRDD = rdd.randomSplit(Array(1.0,3.0,6.0))

#返回的一个RDD数组，查看数组元素

splitRDD(0).collect

splitRDD(1).collect

splitRDD(2).collect

Øglom():RDD[Array[T]]

glom函数是将RDD中每一个分区中类型为T的元素转换为Array[T]

val rdd = sc.makeRDD(1 to 10,3)

val glomRDD= rdd.glom

#返回的结果是一个数组，

glomRDD.collect

scala> glomRDD.collect

res44: Array[Array[Int]] = Array(Array(1, 2, 3), Array(4, 5, 6), Array(7,

8, 9, 10))

1.3.4.3 RDD的集合操作

Øunion(other:RDD[T]) :RDD[T]

将两个RDD的数据进行合并，返回两个RDD的并集，不去重

Øintersection(other:RDD[T]) :RDD[T]

返回两个RDD的交集（会去重）

Øsubtract(other:RDD[T]) :RDD[T]

取差集

Øzip(other:RDD[T]):RDD[T]

zip函数用于将两个RDD组合成key/value形式的RDD，两个RDD的partition个数以及元素的数量都必须要相同，不然会抛出异常

val rdd1 = sc.makeRDD(List(1,2,3,3))

val rdd2 = sc.makeRDD(List(2,3,4))

union操作

rdd1.union(rdd2).collect

#结果

Array[Int] = Array(1, 2, 3, 3, 2, 3, 4)

intersection操作

rdd1.intersection(rdd2).collect

#结果（去重）

Array[Int] = Array(3, 2)

subtract操作

rdd1.subtract(rdd2).collect

#结果（不去重）

Array[Int] = Array(1, 1)

zip操作

scala> val rdd1 = sc.makeRDD(1

to 3)

scala> val rdd2 =

sc.makeRDD(List(1.0,2.0,3.0))

scala> rdd1.zip(rdd2).collect

res8: Array[(Int, Double)] = Array((1,1.0), (2,2.0), (3,3.0))

1.3.4.4键值RDD转换操作

map和flatmap的区别：

（1）map是对每个元素都进行指定的操作，返回每个元素处理后的对象

（2）flatmap对所有的元素都做指定的操作，将所有的对象合并为一个对象返回

val rdd = sc.makeRDD(1 to 3)

rdd.map(x=>Seq(x,x)).collect

#结果

Array[Seq[Int]] = Array(List(1, 1),

List(2, 2), List(3, 3), List(4, 4))

rdd.flatMap(x=>Seq(x,x)).collect

#结果，合并为一个对象返回

Array[Int] = Array(1, 1, 2, 2, 3, 3, 4,  4)

未完待续.....

1.3.5 RDD的行动操作

每调用一次行动操作，都会触发一次spark的调度并返回响应的结果

1.3.5.1集合标量行动操作

Øcount

返回RDD中的元素的个数

Øfirst

返回RDD中的第一个元素

Øreduce(f:(T,T) = >T)

对RDD中的元素进行二元计算，返回计算结果

val rdd = sc.makeRDD(1 to 4)

rdd.reduce(_+_)#10

rdd.reduce(_-_)#-8

Øcollect()

以集合的形式返回RDD的元素

Øtake(number:Int)

返回集合中[0,num-1]下标的元素

Øtop(num:Int)

先降序排序，返回前num个元素

ØtakeOrdered(num:Int)

以与top相反的排序规则（升序），返回前num个元素

Ølookup(key:k):Seq[v]

lookup是针对（k,v）类型RDD的行动操作，针对给定的键值，返回与此键值相对应的所有值

1.3.5.2存储行动操作

RDD不仅可以存储在hdfs中还能存储到Hbase、MangoDB等数据库中

1.4 Spark SQL

1.4.1 spark sql与shark区别

sparksql是一个支持结构化数据处理的spark模块，提供DaraFrame作为可编程的数据抽象，可以对DataFrame执行sql的操作。

spark sql的诞生就是为了解决spark平台上的交互式查询问题，并且提供sql接口兼容原有数据库用户的使用情况

Øshark简单的说，就是spark上的hive，其底层依赖Hive引擎的，但在spark平台上，解析速度是hive的好多倍；就是一个升级版的大数据仓库

Ø在spark1.0版本开始，shark被官方抛弃使用

Øspark sql的优势：

nspark sql完全脱离了hive的限制

nspark sql支持查询原生的RDD，能够高效的处理大数据的各种场景的基础

n能够在scala中写sql语句，支持简单的sql语法检查，将结果取回作为RDD使用

nCatalyst能够帮助用户优化查询，catalyst能够进行一定程度的性能提升

# catalyst是spark sql的调度核心，解析sql形成其对应的执行计划（遵循DAG图）

1.4.2 DataFrame和DataSet

1.4.2.1 RDD与DataFrame的区别

如上图：

Ø左侧的RDD[Person]虽然以Person为类型参数，但是spark框架本省不了解Person类的内部结构；而右侧的DataFrame却提供了详细的结构信息（schema），使得spark sql可以清楚的知道该数据集中包含哪些列，每列的名称和类型

ØRDD是分布式的java对象的集合；而DataFrame是分布式的row对象的集合

ØDataFrame除了提供比RDD更丰富的算子外，更重要的特点是提升执行效率、减少数据读取以及执行计划的优化

1.4.2.2 DataFrame与DataSet的区别

DataSet可以认为是DataFrame的一个特例，主要区别是DataSet的每一个record存储的是一个强类型值而不是一个Row。具有三个特点：

ØDataSet在编译时检查类型

Ø面向对象的编程接口

Ø后面的版本DataFrame是继承DataSet的，DataFrame是面向Spark sql的接口

相互转换：

DataFrame和DataSet可以相互转化，df.as[ElementType]这样可以把DataFrame转化为DataSet，ds.toDF()这样可以把DataSet转化为DataFrame

1.4.2.3 DataFrame

（1）DataFrame是一个分布式的数据集，类似于关系数据库的一个表。

ØDataFrame以列的形式存储，但是不知道列的类型，因此，在编译时不进行校验，只有在运行时才会处理；DataSet不仅知道字段，还知道类型，所以编译时会进行类型校验

Ø可以由结构化的数据转换过来，也可以从hive，外部数据库或者RDD转换

ØDataFrame在spark sql中，可以使用sql的方式进行操作，与RDD类似，也可以采用lazy的方式，只有动作发生时才会真正的计算

ØDataFrame的数据源：支持JSON文件、hive表格，支持本地文件系统以及hdfs等；配合JDBC还支持外部关系型数据库

1.4.2.4与RDD的相互操作

spark sql支持两种不同的方式用于将存在的RDD转换为DataSets、DataFrame

Ø反射推断模式：

该模式使得代码更加的简练，不过在写spark程序的时候已经知道模式信息，（比如RDD中自己定义的case class类型）

练习：从hdfs文件中读取数据，创建一个Person的RDD

1.定义Person type

scala> case class

Person(id:Int,name:String,addr:String)

defined class Person

2.从hdfs读取文件，封装成DataFrame数据集

scala> val personDf = spark.sparkContext

.textFile("/test/preson.txt")

.map(_.split(","))

.map(p=>Person(p(0).toInt,p(1),p(2)))

.toDF

personDf:

org.apache.spark.sql.DataFrame = [id: int, name: string ... 1 more field]

3.将personDf注册为一个视图view

scala>

personDf.createOrReplaceTempView("person")

4.通过sql查询视图；sql支持复杂的,包括多表关联

scala> spark.sql("select

addr,count(1) from person group by addr").show

Ø编程指定模式：

构造一个模式，将其应用到一个已经存在的RDD上将其转化为DataFrame，该方法适用于运行之前不知道列以及列的类型的情况

import org.apache.spark.sql.types._

1.加载数据

val presonRDD= spark.sparkContext.textFile(“/test/person.txt”)

2.定义schema

val stringSchema = “id,name,addr”

val schema = StructType(stringSchma.split(",").map(field=>StructField(field,StringType,nullable=true)))

3.转换rdd的记录到rows集合

1.4.3 spark sql的操作

1.4.3.1创建SparkSession实例

SparkSeesion类时Spark SQL的所有功能的入口；spark-shell启动时，默认生成了一个SparkSession的实例：spark

importorg.apache.spark.sql.SparkSession

val spark = SparkSession.builder

.master("local")

.config("spark.sql.warehouse.dir", "/user/hive/warehouse")

.appName("spark text")

.getOrCreate

//包含隐式转换（比如讲RDDs转成DataFrames）API

importspark.implicits._

1.4.3.2创建DataFrame

spark sql读取hdfs中json数据

val df = spark.read.json("/test/course.json")

#显示df的数据

Ø显示df的数据

df.show

#结果：

Ø查询df的结构信息

Ø显示指定的字段值：使用select(col1,,col2)

Ø过滤，查询长度》12的数据

Ø分组操作groupBy

1.4.3.3 spark sql实战

1.4.3.3.1入口：SQLContext,HiveContext（Starting Point: SQLContext）

spark sql中所有的操作入口点都是SQLContext类或者它的子类.创建一个基本的SQLContext，只需要SparkContext即可（sc）

注：spark2.0之后，sparkSession是实现了同样的功能，不需要显示的创建SparkConf、sparkContext、SQLContext,因为这些对象都封装在了SparkSession中。即是

val sqlContext = new org.apache.spark.sql.SQLContext(sc)

valhiveContext =neworg.apache.spark.sql.hive.HiveContext(sc)

除了SparkContext外，还有HiveContext。两者的区别：

ØSQLContext只支持标准的sql语法解析器

ØHiveContext现在支持sql语法解析器和HiveSql语法解析器；默认为hivesql语法解析器，用户可以通过配置来切换sql语法解析器，来运行hivesql不支持的语法

Ø使用HiveContext可以使用Hive的UDF，读写Hive表数据等hive操作。sqlContext不可以对hive进行操作

Ø趋势：SqlContext不断丰富中，最终两者会形成一个统一的Context

1.4.3.3.2创建DataFrame

使用SqlContext，spark程序可以通过RDD、hive表、JSON格式数据等数据源创建DataFrame

val df = sqlContext.read.json(“/test/readme.json”)

1.4.3.3.3 DataFrame操作

df.show

df.printSchema

df.

1.4.3.3.4 Parquet文件

Parquet文件是一种列式存储格式的文件，能被很多数据处理系统支持。Spark SQL支持读取和写入Parquet文件，并可自动保留原始数据的格式（schema）

优势：

可以跳过不符合条件的数据，只读取需要的数据，降低IO数据量

压缩编码可以降低磁盘存储空间。由于同一列的数据类型是一样的，可以使用更高效的压缩编码进一步节省存储空间

只读取需要的列，支持向量运算，能够获取更好的扫描性能

还有，parquet数据源支持自动发现和推断分区信息

1.4.3.3.5 DataFrame的java操作

将DataFrame的结果转换为java的list

List listRow =result.javaRDD().collect();

for(Row row : listRow){

System.out.println(row);

}

1.4.3.4 SparkSession操作

在2.0版本之前，与spark交互之前必须创建SparkConf和SparkContext；然而到了2.0版本，不需要显示的创建这些对象SparkConf、SparkContext和SqlContext了，这些对象都已经封装在了SparkSession中了，即，2.0版本之后，入口就是就变成了SparkSession，在spark-shell启动时，会实例化一个SparkSession实例spark

ØSparkSession封装的对象

Ø获取conf默认配置，可以调整配置spark的运行参数

1.4.3.4.0 SparkSession的创建

sparkSession类是所有Spark SQL功能的入口，只需要调用SparkSession.builder()即可创建

importorg.apache.spark.sql.SparkSession

val spark = SparkSession.builder

.master("local")

.config("spark.sql.warehouse.dir", "/user/hive/warehouse")

.appName("spark Streaming +kafka")

.enableHiveSupport

.getOrCreate

1.4.3.4.1获取catalog元数据

1.4.3.4.2创建Dataset和Dataframe

最简单的办法就是通过range方法，创建DataSet

注，range也可以有3个参数，第三个参数是间隔，默认的创建的字段：id

Øtop(n)操作

Ø对某一列进行统计操作

Ø通过createDataFrame创建

重新命名列名withColumnRenamed

1.4.3.4.3读取json文件

1.4.3.4.4在SparkSession中使用Spark SQL

1.4.3.4.5数据源

spark支持多种数据源的数据，

Ø最简单的加载方式是load，默认的格式为parquet文件，（可以通过spark.sql.sources.default来默认指定格式）

val df = spark.read.load(“...”)

Ø将DataFrame数据存储为parquet

df.select("name","type").write.save("course.parquet")

存储的路径为（hdfs）：path hdfs://192.168.21.144:9000/user/root/course.parquet

也可以手动指定格式，以及指定要保存的文件的格式

scala> val df =

spark.read.format("json").load("/test/course.json")

df: org.apache.spark.sql.DataFrame = [length: bigint, name: string

... 1 more field]

#指定要保存的文件的格式

scala>

df.write.format("parquet").save("course1.parquet")

1.4.3.4.6保存数据到永久表saveAsTable

DataFrame可以通过调用saveAsTable方法将数据落地到hive表中，不过对已经部署的hive不会受影响，spark会创建本地的metastore（使用derby），saveAsTable会持久化数据并指向hive metastore

saveAsTable默认会创建一个“受管理表”，意味着数据的位置都是受metastore管理的。当“受管理表”被删除，其对应的数据也都会被删除。

注：文件内容保存在${SPARK_HOME}/bin/spark-warehouse/tableName

scala> df.write.saveAsTable("course")

#调用，spark.sql(sql_str)比较灵活

1.4.3.4.7 spark整合hive

如果spark没有整合hive，那么spark的元数据都是在bin目录下，自动创建metastore_db（以derby做支撑）

整合hive后，支持spark从hive取数，永久保存数据到hive中；并支持hive的mysql作为元数据存储数据库

Ø将hive配置文件中hive-site.xml文件复制到${spark_home}/conf

Ø将hadoop配置文件中hdfs-site.xml和core-site.xml文件复制到${spark_home}/conf

Ø将hive下元数据库mysql的驱动，复制到${spark_home}/jars下

可以将DataFrame数据永久保存到hive表中

1.5 Spark Streaming

spark2.0将流数据计算统一到了DataSet中，提出了Structured

Streaming的概念，将数据源映射为一张无限长度的表，同时将流计算的结果映射为另一张表，完全以结构化的方式去操作流数据，复用了其对象的Catalyst引擎

1.5.1 spark Streaming实战

创建Steaming DataFrame，用来监听host:9999获取socket数据，并对获取的数据进行RDD转换操作，最后统计各个词出现的次数

Ø创建socket通道

nc –lk 9999

Ø获取socket通道数据(需要填写socket的host、port)

scala>val line

=

spark.readStream.format("socket").option("host","mapeng").option("port",9999).load

Ø转换操作

scala> val wordCount = line.as[String].flatMap(_.split("

")).groupBy("value").count

wordCount: org.apache.spark.sql.DataFrame

= [value: string, count: bigint]

Ø使用start()来启动流式数据计算流程

scala> val query =

wordCount.writeStream.outputMode("complete").format("console").start

程序自动启动了job计算，并在控制台展现计算结果；（接收到socket流数据，spark自动计算，控制台展现结果）

说明：

（1）outputMode现在有三种方式：complete ,append,update（目前只实现了前两种）

lcomplete：每次计算完成后，都能得到全量的计算结果（每次计算都得到转换后的最新结果集）

lappend：每次计算完成后，能拿到增量的计算结果

两种方式的使用说明：

使用了聚合类函数才能使用complete的模式，只有简单的使用了map,filter等转换模式才能使用append模式，不做复杂的聚合统计运算

1.6 Spark Streaming + kafka整合

1.6.1 pom.xml文件，添加spark依赖

org.apache.hadoop

hadoop-client

2.6.0

org.apache.hadoop

hadoop-common

2.6.0

org.apache.hadoop

hadoop-hdfs

2.6.0

org.apache.spark

spark-core_2.11

2.0.0

org.apache.spark

spark-sql_2.11

2.0.0

org.apache.spark

spark-streaming_2.11

2.0.0

org.apache.spark

spark-hive_2.11

2.0.0

org.apache.spark

spark-streaming-kafka-0-10_2.11

2.0.0

org.apache.hive

hive-jdbc

1.2.1

io.netty

netty-all

4.0.29.Final

1.6.2实战代码

packagecom.mp.fight

importorg.apache.spark.sql.SparkSession

importorg.apache.spark.streaming.StreamingContext

importorg.apache.spark.streaming.Seconds

importorg.apache.spark.streaming._

importorg.apache.spark.streaming.kafka010.KafkaUtils

importorg.apache.spark.streaming.kafka010.LocationStrategies

importorg.apache.kafka.common.serialization.StringDeserializer

importorg.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe

importorg.apache.spark.sql.SQLContext

importorg.apache.spark.sql.SaveMode

objectTest4{

caseclassPerson(id:Int,name:String,addr:String)

defmain(args: Array[String]): Unit = {

//声明sparkSession

valspark=SparkSession.builder

.master("local")

.appName("spark  Streaming kafkasql")

.config("spark.sql.warehouse.dir","/user/hive/warehouse")

.getOrCreate

//kafka设置

valkafkaParams=Map[String, Object](

"bootstrap.servers"->"mapeng:9092",

"key.deserializer"->classOf[StringDeserializer],

"value.deserializer"->classOf[StringDeserializer],

"group.id"->"example",

"auto.offset.reset"->"latest",

"enable.auto.commit"->(false: java.lang.Boolean)

)

//topic

valtopics=List("testmp")

//初始化StreamingContext

valssc=newStreamingContext(spark.sparkContext,Seconds(30));

//从kafka中读取数据

valkafkaStream=KafkaUtils.createDirectStream[String,String](

ssc,

LocationStrategies.PreferConsistent,

Subscribe[String,String](topics,kafkaParams)

).map(_.value())

//kafkaStream.print

importspark.sqlContext.implicits._

//启用sparkSql来操作DStream转换为DataFrame

kafkaStream.foreachRDD{rdd=>{

if(rdd.isEmpty) {

println("rdd is

empty")

}else{

valperson=rdd.map(_.split(",")).map(p=>Person(p(0).toInt,p(1),p(2))).toDF

//新接收的数据，追加存储在parquet文件中（重写文件）

person.write.mode(SaveMode.Append).save("hdfs://mapeng:9000/test/person.parquet")

//实时统计区域人数

valdf=spark.read.load("hdfs://mapeng:9000/test/person.parquet")

df.createOrReplaceTempView("person")

valaddrCount=spark.sql("select

addr,count(1) as num from person group by addr")

//将统计结果实时回写到parquet文件中

addrCount.write.mode(SaveMode.Overwrite).save("hdfs://mapeng:9000/test/addrCount.parquet")

//继续做多维度统计，可以使用sparksql操作处理parquet文件

}

}

}

//启动job

ssc.start

ssc.awaitTermination

}

}

1.6.3 DStream中foreachRDD、foreachePartition、foreach的区别

ØforeachRDD：得到的是处理一个批次的数据

ØforeachPartition：对一个批次的每个分区数据做处理

Øforeach：每条数据处理，单个元素处理

1.6.4 spark Streaming + socket

val ssc = new StreamingContext(sparkConf,

Seconds(1))

//获得一个DStream负责连接监听端口:地址

val lines = ssc.socketTextStream(“192.168.21.144”, 9999)

//对每一行数据执行Split操作

val words = lines.flatMap(_.split(" "))

//统计word的数量

val pairs = words.map(word => (word, 1))

val wordCounts = pairs.reduceByKey(_ + _)

//输出结果

wordCounts.print

ssc.start//开始

ssc.awaitTermination//计算完毕退出

1.8 parquet文件

1.8.1 parquet是面向分析型业务的列式存储格式，有如下优势

Parquet文件尾部存储了文件的元数据信息和统计信息，自描述的，方便解析

1.只读取需要的列，支持向量运算，能够获得更好的扫描性能

2.可以跳过不符合条件的数据，只读取需要的数据，降低io

3.同一列的数据类型是一样的，可以使用更高效的压缩编码，节约存储磁盘

1.8.2 parquet适配多种计算框架

Parquet是语言无关的，而且不与任何一种数据处理框架绑定在一起，适配多种语言和组件，能够与Parquet配合的组件有：

查询引擎: Hive, Impala, Pig, Presto, Drill, Tajo, HAWQ, IBM

Big SQL

计算框架: MapReduce, Spark, Cascading, Crunch, Scalding, Kite

数据模型: Avro, Thrift, Protocol Buffers, POJOs

1.8.3 parquet数据模型

eg:

message AddressBook {

required string owner;

repeated stringownerPhoneNumbers;

repeated group contacts {

required string name;

optional string phoneNumber;

}

}

说明：

schema的格式是这样的：

（1）根叫做message

（2）message包括有多个fields，每个fields包括有三个属性：repetition，type，name

其中，repetition有3中类型：required（出现一次）、optional（出现0次或者1次）、repeated（出现0次或者多次）

（3）type可以是一个group或者一个简单的类型

以上schema描述说明：

（1）每条记录标识一个AddressBook

（2）有且只有一个owner

（3）有0个或多个ownerPhoneNumbers

（4）owner可以有0个或者多个contacts。每个contact有且只有一个name，这个contact的phoneNumber可有可无（0个或者1个）

注：parquet格式的数据类型没有复杂的Map，List，Set等，使用group和repeated fields来表示；

null值不会被存储

实例：

1.8.4 parquet格式文件的存储

在parquet格式的存储中，一个schema的树结构有几个叶子节点，实际存储汇总就有几个column，例如上图中的schema实际存储就4个列

1.8.5 DataFrame与Parquet

（1）保存DF为Parquet格式

dfPerson.write.parquet("person.parquet")

(2)hive中建立parquet格式的表

createtableperson_parquetlikepersonstoredasparquet;

insertoverwritetableperson_parquetselect*fromperson;

（3）加载Parquet文件不再需要case class。

valpersonDF =spark.read.parquet("person.parquet")

personDF.registerAsTempTable("pp")

valmales = spark.sql("select * from pp where gender='M'")

males.show

1.8.5 parquet文件的持久化

1.8.5.1 spark中将DataFrame数据写到hdfs中的parquet文件中，支持追加

personDf.write.mode(SaveMode.Append).save(“hdfs://mapeng:9000/test/person.parquet”)

saveMode有如下几种方式：

1.8.5.2 parquet文件合并

合并的规则：相同的列，在新的数据集中，是通用的列，

各自不同的列，也作为新的数据集的列。

实战：

1.9 spark项目实战