Spark 程序设计

清风云悦
440 阅读5分钟

一、Spark Shell on Client

scala> var rdd =sc.parallelize(1 to 100 ,3)
rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[0] at parallelize at <console>:24
scala> rdd.count
res0: Long = 100  
scala> val rdd2=rdd.map(_ + 1)
rdd2: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[1] at map at <console>:26
scala> rdd2.take(3)
res1: Array[Int] = Array(2, 3, 4)
scala> val rdd1=sc.textFile("file://home/hadoop/apps/sparkwc")
rdd1: org.apache.spark.rdd.RDD[String] = file://home/hadoop/apps/sparkwc MapPartitionsRDD[3] at textFile at <console>:24
cala> val rdd1=sc.textFile("file:///home/hadoop/apps/sparkwc")
rdd1: org.apache.spark.rdd.RDD[String] = file:///home/hadoop/apps/sparkwc MapPartitionsRDD[9] at textFile at <console>:24

scala> val rdd2=rdd
rdd   rdd1   rdd2   rdd3   rddToDatasetHolder

scala> val rdd2=rdd1.flatMap(_.split("\t"))
rdd2: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[10] at flatMap at <console>:26

scala> val rdd3=rdd2.map((_,1))
rdd3: org.apache.spark.rdd.RDD[(String, Int)] = MapPartitionsRDD[11] at map at <console>:28

scala> val rdd4=rdd3.reduceByKey(_ + _)
rdd4: org.apache.spark.rdd.RDD[(String, Int)] = ShuffledRDD[12] at reduceByKey at <console>:30

scala> rdd4.collect
res2: Array[(String, Int)] = Array((spark,1), (hadoop,1), (hello,3), (world,1))
scala> rdd4.collect
res2: Array[(String, Int)] = Array((spark,1), (hadoop,1), (hello,3), (world,1))

scala> rdd4.saveAsTextFile("file:///home/hadoop/apps/out1")
[hadoop@hadoop01 apps]$ cd out1/
[hadoop@hadoop01 out1]$ ls
part-00000  _SUCCESS
[hadoop@hadoop01 out1]$ cat part-00000 
(spark,1)
(hadoop,1)
(hello,3)
(world,1)
[hadoop@hadoop01 out1]$ pwd
/home/hadoop/apps/out1

WebUI 地址：http://192.168.43.20:4040/jobs/

二、Spark Shuffle

  • Shuffle Write：将Task中间结果数据写入到本地磁盘
  • Shuffle Read：从Shuffle Write阶段拉取数据到内存中并行计算
    SparkShuffle Write

三、Shuffle Write（hash-based）

  • Shuffle Write阶段产生的总文件数=MapTaskNum * ReduceTaskNum
  • TotalBufferSize=CoreNum * ReducceTaskNum*FileBufferSize
  • 产生大量小文件，占用更多的内存缓冲区，造成不必要的内存开销，增加 了磁盘IO和网络开销
    Shuffle Write

四、Shuffle Write（hash-based优化）

  • Shuffle Write阶段产生的总文件数=CoreNum * ReduceTaskNum
  • TotalBufferSize=CoreNum * ReducceTaskNum*FileBufferSize 减少了小文件产生的个数，但是占用内存缓冲区的大小没变
  • 设置方法
    • conf.set("spark.shuffle.manager", "hash")
    • 在conf/spark-default.conf 配置文件中添加spark.shuffle.manager=hash
      Shuffle Write优化

五、Shuffle Write（hash-based优化）Shuffle Write（sort-based）

  • Shuffle Write阶段产生的总文件数= MapTaskNum * 2
  • 优点： 顺序读写能够大幅提高磁盘IO性能，不会产生过多小文件，降低文件缓存占用内存空间大小，提高内存使用率。
  • 缺点：多了一次粗粒度的排序。
  • 设置方法
  • 代码中设置：conf.set("spark.shuffle.manager", "sort")
  • 在conf/spark-default.conf 配置文件中添加spark.shuffle.manager=sort
    sort-based

六、Shuffle Read

  • hase-based和sort-based使用相同的shuffle read实现
    Shuffle Read

七、Spark History Server配置

  • spark history server查看运行完成的作业信息和日志
  • 配置Hadoop的yarn-site.xml文件，所有节点配置文件同步，重启yarn
<property>
<name>yarn.log.server.url</name>
<value>http://node02:19888/jobhistory/logs</value>
<description> Yarn JobHistoryServer访问地址 </description>
</property>
  • 修改spark安装包conf目录下的spark-defaults.conf（如果没有该文件， 通过spark-defaults.conf.template模板复制一个），spark history server 在192.168.183.100节点启动，spark_logs这个目录需要在HDFS上提前创建
spark.yarn.historyServer.address=192.168.183.100:18080  spark.history.ui.port=18080
spark.eventLog.enabled=true  spark.eventLog.dir=hdfs:///spark_logs
spark.history.fs.logDirectory=hdfs:///spark_logs

1.Spark History Server启动

  • 启动Spark History Server
sbin/start-history-server.sh
  • Spark History Server访问地址
httpL://192.168.183.100:18080

Spark history

七、Spark运行环境优化

  • 将spark系统jar包上传到HDFS上，直接使用HDFS上的文件
  • 在spark安装目录下运行：jar cv0f spark-libs.jar -C jars/ .
  • 将spark安装目录下生成的spark-libs.jar上传到HDFS上的 /system/spark（需要手动创建）目录下
hadoop fs -put spark-libs.jar /system/spark

修改spark安装包conf目录下spark-defaults.conf配置文件添加spark- libs.jar在HDFS上的路径

spark.yarn.archive=hdfs:///system/spark/spark-libs.jar

八、Spark编程模型

  • 创建SparkContext
    • 封装了spark执行环境信息
  • 创建RDD
    • 可以用scala集合或hadoop数据文件创建
  • 在RDD上进行transformation和action
    • spark提供了丰富的transformation和action算子
  • 返回结果
    • 保存到hdfs、其他外部存储、直接打印

1.提交Spark程序到Yarn上

2.Spark RDD算子分类

  • Transformation转换操作，惰性执行，不触发app执行
    • 针对Value数据类型，如map、filter
    • 针对Key-Value数据类型，如groupByKey、reduceByKey
  • Action执行操作，触发app执行

3.创建RDD

  • parallelize从集合创建RDD
    • 参数1：Seq集合，必须
    • 参数2：分区数
    • 创建RDD：val rdd = sc. parallelize(List(1,2,3,4,5,6,7),3)
    • 查看RDD分区数：rdd.partitions.size
  • textFile从外部数据源（本地文件或者HDFS数据集）创建RDD
    • 参数1：外部数据源路径，必须
    • 参数2：最小分区数
    • 从本地文件创建RDD：val rdd = sc.textFile("file:///home/hadoop/apps/in")
    • 从HDFS数据集创建RDD：val rdd = sc.textFile("hdfs:///data/wc/in",1)

4.Value数据类型Transformation

  • map
    • 输入是一个RDD，将一个RDD中的每个数据项，通过map中的函数映射输出一个新的RDD，输入分区与输出分区一一对应
  • flatMap
    • 与map算子功能类似，可以将嵌套类型数据拆开展平
  • distinct
    • 对RDD元素进行去重
  • coalesce
    • 对RDD进行重分区
    • 第一个参数为重分区的数目
    • 第二个为是否进行shuffle，默认为false，如果重分区之后分区数目大于 原RDD的分区数，则必须设置为true
  • repartition
    • 对RDD进行重分区， 等价于coalesce第二个参数设置为true
  • union
    • 将两个RDD进行合并，不去重
  • mapPartitions
    • 针对RDD的每个分区进行操作，接收一个能够处理迭代器的函数作为参数
    • 如果RDD处理的过程中，需要频繁的创建额外对象，使用mapPartitions要比使用map的性能高很多，如：创建数据库连接
  • mapPartitionsWithIndex
    • 与mapPartitions功能类似，接收一个第一个参数是分区索引，第二个参数是分区迭代器的函数
  • zip
    • 拉链操作，将两个RDD组合成Key-Value形式的RDD，保证两个RDD的partition数量和元素个数要相同，否则会抛出异常
  • mapValues
    • 针对[K,V]中的V值进行map操作
  • groupByKdy
    • 将RDD[K,V]中每个K对应的V值，合并到一个集合Iterable[V]中
  • reduceByKey
    • 将RDD[K,V]中每个K对应的V值根据传入的映射函数计算
  • join -返回两个RDD根据K可以关联上的结果，join只能用于两个RDD之间的关联，如果要多个RDD关联，需要关联多次

5.RDD Action

  • collect
    • 将一个RDD转换成数组，常用于调试
  • saveAsTextFile
    • 用于将RDD以文本文件的格式存储到文件系统中
  • take
    • 根据传入参数返回RDD的指定个数元素
  • count
    • 返回RDD中元素数量

6.Spark优化-Cache应用

Cache应用

7.Accumulator计数器

  • accumulator累加器，计数器
    • accumulator累加器，计数器
    • 通常用于监控，调试，记录关键数据处理的数目等
    • 分布式计数器，在Driver端汇总
val total_counter = sc.accumulator(0L,"total_counter")  
val resultRdd = rowRdd.flatMap(_.split("\t")).map(x=>{  total_counter += 1
(x,1)
}).reduceByKey(_ + _)

通过Spark Web UI查看

Accumulator计数器

avatar
文章
阅读
粉丝
相关推荐
一文初探spark
640阅读
 · 
3点赞
Spark机器学习库MLlib编程实践
1.0k阅读
 · 
0点赞
Spark Streaming编程初级实践
277阅读
 · 
0点赞
Spark SQL 编程初级实践
811阅读
 · 
1点赞
浅谈Spark分布式计算
1.1k阅读
 · 
3点赞