详谈Spark的shuffle过程

1、 详谈Spark的shuffle过程shuffle操作，是在Spark操作中调用了一些特俗的算子才会触发的一种操作，因此会导致大量的数据在不同的节点之间传输，所以shuffle过程是spark中最复杂、最消耗性能的一种操作。如reducerByKey算子会将上一个RDD中的每一个key对应的value都聚合成一个 value，然后生成一个新的RDD，新的RDD的元素类型就是的格式，每个key 对应一个聚合起来的value。这里会出现一个问题。对于上一个RDD来说，不是一个key对应的所有value都在一个partition中，也不是key的所有value都在一个节点上。对于这种情况，就必须在集

2、群中将各个节点上同一个 key对应的value同一传输到一个节点上，这个过程会发生大量的网络IO。Shuffle过程分为：shuffle write 和shuffle read 。并且会在不同的stage中进行。在进行一个key对应的values的聚合时，首先，上一个stage的每个map task就必须保证将自己处理的当前分区中的数据相同的key写入一个分区文件中，可能会多个不同的分区文件。接着下一个stage的reduce task就必须从上一个stage的所有task所在的节点上，从各个task写入的多个分区文件中找到属于自己的分区文件，然后将属于自己的分区数据拉取过来，这样就可以保证每

3、个key对应的所有values都汇聚到一个节点上进行处理和聚合，这个过程就称为shuffle过程。shuffle过程中的分区排序：默认情况下，shuffle操作是不会对每个分区中的数据进行排序的如果想排序，可以使用三种方法：1.使用mapPartitions算子。2.使用repartitionAndSortWithinPartitions，该算子是对RDD进行重分区的算子，在重分区的过程中可以实现排序。3.使用sortByKey对所有分区的数据进行全局排序。以上三种方法，mapPartitions代价比较小，因为不需要进行额外的shuffle操作，repartitionAndSortWithi

4、nPartitions和sortByKey可能会进行额外的shuffle操作，因此性能并不是很高。会发生shuffle的算子：1.byKey类的算子：如reduceByKey、groupByKey、sortByKey、aggregateByKey、combineByKey2.repartition类的算子：如repartition（少量分区变成多个分区会发生shuffle）、repartitionAndSortWithinPartitions、coalesce（需要指定是否发生shuffle）、 partitionBy3.join类的算子：如join（先groupByKey后再join就不会发

5、生shuffle）、cogroup，这里需要注意，对于上述操作，能不使用shuffle操作，就尽量不用，尽量使用不发生shuffle的操作。另外，如果使用了shuffle操作，那么必须要进行shuffle的调优，甚至是解决遇到的数据倾斜问题。为了实施shuffle操作，spark才有了stage的概念，在发生shuffle操作的算子中，需要进行stage的划分 shuffle操作的前半部分，属于上一个stage的范围，通常称之为map task， shuffle操作的后半部分，属于下一个stage的范围，通常称之为reduce task，其中map task负责数据的组织，也就是将同一个key

6、对应的value都写入同一个下游task对应的分区文件中，其中reduce task负责数据的聚合，也就是将上一个stage的task所在的节点上，将属于自己的各个分区文件都拉取过来进行聚合，map task会将数据先保存在内存中，如果内存不够时，就溢写到磁盘文件中，reduce task 会读取各个节点上属于自己的分区磁盘文件到自己节点的内存中进行聚合。 shuffle过程中，会产生大量的中间文件，也就是map side写入的大量分区文件，这些文件会一直保留着，直到RDD不再被使用，而且被gc回收掉了，才会去清理中间文件，这样操作主要是为了如果要重新计算shuffle后RDD，那么map s

7、ide不需要重新再做一次磁盘写操作，但是这种情况下，如果在应用程序中一直保留着对RDD的引用，导致很长的时间以后才会进行回收操作，保存中间文件的目录，由spark.local.dir属性指定 spark性能的消耗体并现在在内存的消耗，磁盘IO，网络的IO之中。此外，这里和大家简要介绍下spark2x，以DataSet为操作。DataSet案例一：DataFrame转化DataSet的两种方法：object SparkSQLDemo2 def main(args: ArrayString): Unit = val spark = SparkSession .builder() .appName(

8、Spaek) .master(local2) .getOrCreate() import spark.implicits._ /读取数据 val df: DataFrame = spark.read.json(E:/hadoop/spark阶段/sparkcoursesinfo/spark/spark2xdata/people.json) /把DataFrame构建成DataSet /方法一/ val personDS: DatasetPerson = Seq(Person(dazhao,38,99).toDS()/ personDS.show() /方法二 val personDS: Dat

9、asetPerson = df.asPerson personDS.show() spark.stop() case class Person(name: String, age: Long, facevalue: Long)hive创建和基本开发object SparkSQLDemo3 def main(args: ArrayString): Unit = val spark = SparkSession .builder() .appName(spqrk) .master(local2) /设置sparksql的元数据仓库的目录 .config(spark.sql.warehouse.di

10、r,e:/spark-warehouse) /启动hive支持 .enableHiveSupport() .getOrCreate() /创建并操作hive表 /spark.sql(create table if not exists src(key int,value string) /生成数据 / spark.sql(load data inpath E:/hadoop/spark段/sparkcoursesinfo/spark/spark2xdata/kv1.txt into table src) spark.sql(select * from src).show() spark.sql

11、(select count(*) from src).show()/ val sqlHiveDF: DataFrame = spark.sql(select key,value from src where key skey:$key,value:$value/ / sqlHiveDS.show() spark.stop() 创建临时表的案例object BasicOperation1 def main(args: ArrayString): Unit = val spark = SparkSession.builder().appName(123).master(local2).getOrC

12、reate() val employee: DataFrame = spark.read.json(E:/hadoop/spark阶段/sparkcoursesinfo/spark/spark2xdata/employee.json) val employeeCached = employee.cache() /创建临时表 employee.createOrReplaceTempView(t_employee) val employeeDF: DataFrame = spark.sql(select * from t_employee where age 35) /写入到本地和HDFS emp

13、loyeeDF.write.mode(append).csv(e:/out-2) /DataSet转化为DataFrame/ val employeeDS: DatasetEmpolyee = employee.asEmpolyee / val employeeDF1: DataFrame = employeeDS.toDF() spark.stop() case class Empolyee(name:String,age:Long,depId:Long,gender:String,salary:Double)案例：统计部门的平均薪资和平均年龄需求:1.只统计年龄在20以上的。2.根据部门名

14、称和员工性别为粒度进行统计。3.统计每个部门的性别的平均薪资和平均年龄。object DepartmentAvgSalaryAndAgeDemo def main(args: ArrayString): Unit = val spark = SparkSession .builder() .appName(123) .master(local2) .getOrCreate() /导入spark sql的上下文 import spark.implicits._ /导入spark sql 的function import org.apache.spark.sql.functions._ /加载数据

15、 val employee: DataFrame = spark.read.json(E:/hadoop/spark阶段/sparkcoursesinfo/spark/spark2xdata/employee.json) val department: DataFrame = spark.read.json(E:/hadoop/spark阶段/sparkcoursesinfo/spark/spark2xdata/department.json) /开始统计 employee /对员工信息进行过滤，只统计大于20岁的员工 .filter(age20) /根据需求，必须把部门信息和员工信息join

16、在一起，注意： / untyped join 两个表的字段连接需要使用 = .join(department,$depId = $id) /根据部门名称和员工性别进行分组 .groupBy(department(name),employee(gender) /最后进行聚合 .agg(avg(employee(salary),avg(employee(age) .show() spark.stop() 结果如下：用流式统计单词个数object StructuredStreamingWC def main(args: ArrayString): Unit = val spark = SparkSe

17、ssion .builder() .appName(123) .master(local) .getOrCreate() import spark.implicits._ /设置一些基本信息 val lines: DataFrame = spark .readStream .format(socket) .option(host,hadoop01) .option(port,6666) .load() / DataFrame转Dataset val linesDS: DatasetString = lines.asString val words: DatasetString = linesD

18、S.flatMap(_.split( ) val sumed: DataFrame = words.groupBy(value).count() val query = sumed .writeStream .outputMode(complete) .format(console) .start() /线程等待 query.awaitTermination() 基本练习：重新分区：reparation，coalesceobject TypeOperation1 def main(args: ArrayString): Unit = val spark = SparkSession .buil

19、der() .appName(123) .master(local2) .getOrCreate() val employee: DataFrame = spark.read.json(E:/hadoop/spark阶段/sparkcoursesinfo/spark/spark2xdata/employee.json) println(初始分区数为：+employee.rdd.partitions.size) /reparation可以少量分区改为多个分区会发生shuffle val employeeRepartitions: DatasetRow = employee.repartition

20、(3) println(调用reparation后的分区数量为：+employeeRepartitions.rdd.partitions.size) /coalesce可以少量分区改为多个分区不会发生shuffle val employeeCoaleace: DatasetRow = employee.coalesce(3) println(调用coalesce后的分区+employeeCoaleace.rdd.partitions.size) spark.stop() 结果：基本练习：去重distinct，dropDuplicates。object TypeOperation2 def ma

21、in(args: ArrayString): Unit = val spark = SparkSession .builder() .appName(123) .master(local2) .getOrCreate() val employee: DataFrame = spark.read.json(E:/hadoop/spark阶段/sparkcoursesinfo/spark/spark2xdata/employee.json) /distinct对于整条数据进行去重 val distinctEmployee: DatasetRow = employee.distinct() dist

22、inctEmployee.show() /dropDuplicates某个字段去重 employee.dropDuplicates(Array(name).show() spark.stop() 基本练习：过滤filter except intersect。object TypeOperation3 def main(args: ArrayString): Unit = val spark = SparkSession .builder() .appName(123) .master(local2) .getOrCreate() import spark.implicits._ val emp

23、loyee1: DataFrame = spark.read.json(E:/hadoop/spark阶段/sparkcoursesinfo/spark/spark2xdata/employee.json) val employee2: DataFrame = spark.read.json(E:/hadoop/spark阶段/sparkcoursesinfo/spark/spark2xdata/employee2.json) val empolyeeDS1 = employee1.asEmpolyee val empolyeeDS2 = employee2.asEmpolyee /filte

24、r：根据自己的逻辑，如果返回true，就保留该元素，否则过滤调 empolyeeDS1.filter(employee = employee.age 30).show() /except：获取到当前DataSet中有但是在另外一个DataSet中没有的元素 empolyeeDS1.except(empolyeeDS2).show() /intersect 获取交集 empolyeeDS1.intersect(empolyeeDS2).show() spark.stop() 基本练习：连接 joinWith、sort。object TypeOperation4 def main(args: Ar

25、rayString): Unit = val spark = SparkSession .builder() .appName(123) .master(local2) .getOrCreate() import spark.implicits._ val employee1: DataFrame = spark.read.json(E:/hadoop/spark阶段/sparkcoursesinfo/spark/spark2xdata/employee.json) val employee2: DataFrame = spark.read.json(E:/hadoop/spark阶段/spa

26、rkcoursesinfo/spark/spark2xdata/employee2.json) val department: DataFrame = spark.read.json(E:/hadoop/spark阶段/sparkcoursesinfo/spark/spark2xdata/department.json) val empolyeeDS1 = employee1.asEmpolyee val empolyeeDS2 = employee2.asEmpolyee val departmentDS = department.asDepartment /join需要指定哪个join v

27、al joined: Dataset(Empolyee, Department) = empolyeeDS1.joinWith(departmentDS,$depId =$id) joined.show() /sort:排序，需要指定用哪个字段排序 empolyeeDS1.sort($salary.desc).show() spark.stop() case class Department(id:Long,name:String)基本练习：randomSplit、sample。object TypeOperation5 def main(args: ArrayString): Unit =

28、val spark = SparkSession .builder() .appName(123) .master(local2) .getOrCreate() import spark.implicits._ val employee1: DataFrame = spark.read.json(E:/hadoop/spark阶段/sparkcoursesinfo/spark/spark2xdata/employee.json) val employeeDS: DatasetEmpolyee = employee1.asEmpolyee / randomSplit根据权重将一个DataSet分

29、成多个DataSet / 第一个参数代表权重，需要一个Array，权重高的DataSet，划分到的数据数量大的几率就大一些 / 第二个参数代表随机种子，基本可以忽略val employeeDSArr: ArrayDatasetEmpolyee = employeeDS.randomSplit(Array(0.2, 0.8) employeeDSArr.foreach(ds = ds.show() / sample：按照一定比例随机抽取数据 / 第一个参数：建立不同采样器：false是伯努利分布（元素可以多次采样），true是泊松分布 / 第二个参数：采样比例 / employeeDS.samp

30、le(false, 0.9).show() spark.stop() 基本练习：avg sum max min count countDistinct。object TypeOperation6 def main(args: ArrayString): Unit = val spark = SparkSession .builder() .appName(123) .master(local2) .getOrCreate() import spark.implicits._ import org.apache.spark.sql.functions._ val employee: DataFr

31、ame = spark.read.json(E:/hadoop/spark阶段/sparkcoursesinfo/spark/spark2xdata/employee.json) val department: DataFrame = spark.read.json(E:/hadoop/spark阶段/sparkcoursesinfo/spark/spark2xdata/department.json) employee.join(department,$depId = $id) .groupBy(department(name) .agg(avg(employee(salary),sum(e

32、mployee(salary), max(employee(salary),min(employee(salary), count(employee(name),countDistinct(employee(name) ).show() spark.stop() 基本练习：1.日期函数：current_date 、current_timestamp2.数学函数：round 保留几位小数3.随机函数：rand4.字符串函数：cancat、cancat_wsobject TypeOperation8 def main(args: ArrayString): Unit = val spark = S

33、parkSession .builder() .appName(123) .master(local2) .getOrCreate() import org.apache.spark.sql.functions._ val employee = spark.read.json(E:/hadoop/spark阶段/sparkcoursesinfo/spark/spark2xdata/employee.json) employee .select( employee(name),current_date(),current_timestamp(), rand(), round(employee(s

34、alary),2), concat(employee(gender),employee(age), concat_ws(|,employee(gender),employee(age) ).show() spark.stop() Redic介绍：Redic支持的类型：字符串类型，散列类型，列表类型，集合类型。Redic安装：1.安装c语言环境yum install -y gcc-c+2.进入安装包，执行make命令，编译Redis的源码cd /usr/local/redis-3.0.0 make具体解析安装：rootredis01 redis-3.0.0# make install PREFI

35、X=/usr/local/redis后端启动第一步：执行cp命令将redis解压缩包中的redis.conf文件拷贝到bin目录下。第二步：修改redis.conf文件：第三步：启动redis服务./redis-server redis.conf启动客户端./redis-cliRedis数据类型String、 Hash、 List 、Set 、SortedSetString操作 ：Set赋值 sets11 get取值 get s1del删除 dels1 自增 value必须为数字类型incr s1 自减 decr s1 自增指定数值 incrbys1 5 自减指定数值 decrbys1 5 设

36、置多个值 msets1 3 s2 4 获取多个值mgets1 s2Hsah操作：hset设置值 hset user age 3 hget取值 hget user agehdel删除 hdel user 同时增加多个值 hmsetuser age 3 name hh 同时获取多个值 hmget user age name 增加数值 hincrby user age 1 判断字段是否存在hexists user age 当字段不存在是赋值，如果存在不执行操作hexists user age 30赋值多个字段 hmset user age 4 name hh 获取user的所有字段hkeys use

37、r 获取user的所有字段值hval user 获取字段数量hlen userList操作：list类型存储采用linkedlist 可以实现队列和堆栈 从左面存值（堆栈）lpish list1 1 2 3 4 5 6从右面存值（队列）rpush list1 a b c d 查看list lrange list 0 3 查看全部 lrange list1 0 -1弹出堆栈的lpop list1 rpop list1 获取列表的长度 llen list1 删除lrem 向列表中插入元素 linsert list1 5 5 将元素从一个列表中转移到另一个列表中rpoplpush list1 lis

38、t2Set类型：set去重无序添加元素sadd sad set1 1 1 3 4 5 删除元素srem set1 3 查看元素smembers set1 判断元素是否存在sismeneber set1 6 集合运算 差集sdiff，交集sinter，并集sunion 获取集合中元素的个数scard set1 从集合中弹出一个元素spop set sortedset类型：添加元素zadd set1 1 haha 2 hehe 0 heihei删除元素zrem zset1 1 haha获得排名在某个范围的元素列表Zrangge set1 0 3 zrevrange zset1 0 3 withso

39、res 增加某个元素的分数，返回值是更改后的分数 zincrby zset1 5 hehe获得集合中元素的数量 zcard zset1 获得指定分数范围内元素的个数zcount zset1 min max 按着排名范围删除元素zremrangebyrank zset1 0 1 按着分数范围删除元素zremrangebyrank zset1 20 60 获取元素排名(从小到大)zrank key member 获取元素排名(从大到小) zrevrank key memberKeys命令查看所有的key key*expire key seconds /设置key的生存时间ttl key /查看ke

40、y生存时间persist key /清空生存时间keys my* 返回满足条件的所有key 例如返回 my1 my2exists key /判断key是否存在delkey 删除key remame key 重命名ket type key返回key的类型Redis持久化方式一 ：rdb方式RDB方式的持久化是通过快照完成的，当符合一定条件是Redis会自动将内存中的数据进行快照并持久化到硬盘。RDB是默认采用的持久化方式，在redis.conf有如下配置：save 开头的一行就是持久化配置，可以配置多个条件（每行配置一个条件），每个条件之间是“或”的关系，“save 900 1”表示15分钟（9

41、00秒钟）内至少1个键被更改则进行快照，“save 300 10”表示5分钟（300秒）内至少10个键被更改则进行快照。其缺点在于，一旦Redis异常退出，就会丢失最后一次快照的数据方式二： aof方式 aof是默认不开启的，需要手动设置。这里注意，需要重新启动服务。Redis集群的搭建1.安装rub yum -y install ruby yum -y install rubygems 2.安装ruby和Redis的解口程序，上传redis3.0.0.gem至/usr/local下，执行gem install /usr/local/redis-3.0.0.gem3.新建一个文件夹 redis

42、-cluster。赋值两份redis 到redis-cluster 取名为 redis01 reids02cd redis01/bin/redis/conf。修改端口号 分别为 6001 6002 6003 6004 6005 6006。打开 cluster-enabled yes 4.cd redis-3.0.0/src。ll *.rb。cp redis-trib.rb /usr/local/redis-cluster5.启动集群。redis/bin/redis-server redis.conf6.关联节点cd /usr/local/redis-cluster/。./redistrib.r

43、b create replicas 1 hadoop01:6001 hadoop01:6002 hadoop02:6003 hadoop02:6004 hadoop03:6005 hadoop03:6006 redisAPI练习Lucene 全文检索工具包Elasticsearch是基于Lucene的搜索服务器，用于存储和全文检索 Solr是搜索服务是windows下安装ElasticSearch 工具包：elasticsearch-2.4.0.zip 安装elasticsearch的前提：安装好jdk 。1.解压elasticsearch-2.4.0.zip。2.进入elasticsearc

44、h/bin 运行elasticsearch.bat范文 http:/localhost:9200 看服务是否启动成功。将Elasticsearch注册到windows服务器 不用每次启动Elaticsearch。注册过程中改变jdk的绝对路径。安装Elasticsearch 的插件elasticsearch-head(用于可视化显示)1)解压 elasticsearch-head-master.zip 更改文件名为head 拷贝进到plugins2)重启服务3)访问节点 http:/localhost:9200/_plugin/head/4)结果显示如下安装分词器 elasticsearch-

45、analysis-ik-1.10.4.zip解压包，更改名为analysis-ik 拷贝到 plugins目录注意：windows下用分词器，需要index和映射关联。ElasticSearch与MySql的对比：ES java操作客户端API idea 程序要仔细查看，linux下安装elasticsearch。因为es需要普通用户启动，所以需要添加普通用户。1.创建bigdata用户 useradd bigdata 为bigdata用户添加密码 echo | passwd -stdin bigdata2.在根目录下创建目录 mkdir /bigdata,data 修改目录的权限chown

46、-R bigdata:bigdata bigdata,data3.上传es包，解压tar -zxvf elasticsearch-2.4.4.tar.gz -C /bigdata/ cd elasticsearch-2.4.4 mkdir plugins4.安装可视化插件/bin/plugin install file:/root/elasticsearch-head-master.zip 5安装分词器/bin/plugin install file:/root/elasticsearch-analysis-ik-1.10.4.zip6.修改配置cd /bigdata/elasticsearc

47、h-2.4.4/config vi elasticsearch.yml修改 和network.host7.分发scp -r elasticsearch-2.4.4/ hadoop02:$PWD scp -r elasticsearch-2.4.4/ hadoop03:$PWD8.用浏览器访问es机器所在9200端口 http:/hadoop01:9200/ 访问head管理页面1:9200/_plugin/headLinux 下Logstash 安装和使用Logstash 是类似于flume ，可以从多种文件格式获取数据，并输出各种格式安装

48、：1.tar -zxvf logstash-2.3.1.tar.gz -C /bigdata/2.启动 第一种方式控制台-控制台（元数据输出）bin/logstash -e input stdin output stdout 控制台-控制台（增加数据格式）bin/logstash -e input stdin output stdoutcodec = rubydebug 控制台-es,控制台bin/logstash -e input stdin output elasticsearch hosts = hadoop01:9200 stdout bin/logstash -e input std

49、in output elasticsearch hosts = hadoop01:9200, hadoop02:9200 stdout 控制台-kafka，控制台bin/logstash -e input stdin output kafka topic_id = test1 bootstrap_servers = hadoop01:9092,hadoop02:9092,hadoop03:9092 stdoutcodec = rubydebug 第二种方式：以文件的形式 input：输入源，可以有多个 filter：数据清洗 可以清理多个 output：输出源，可以有多个可以加多个源 启动形式“：bin/logstash -f 文件名文件-kafka:path表示输入的路径，topid表示kafka的topid Input file codec = plain charset = UTF-8 path = /root/logserver/gamelog.txt discover_interval = 5 start