大数据进阶之路——SparkSQL补充

手写 WordCount

使用flatMap、reduceByKey 来计算

//sc是SparkContext对象&＃xff0c;该对象是提交spark程序的入口
sc.textFile("file:///home/hadoop/data/hello.txt") // 读取文件&＃xff0c;.flatMap(line &＃61;> line.split(" ")) // 将文件中的每一行单词按照分隔符&＃xff08;这里是空格&＃xff09;分隔.map(word &＃61;> (word,1)) //给每个单词计数为1.reduceByKey((x,y) &＃61;> (x&＃43;y)) // 统计相同单词的数量.collect


简写

sc.textFile("file:///home/hadoop/data/hello.txt").flatMap(_.split(" ")).map((_, 1)).reduceByKey(_ &＃43; _).collect

RDD、DAG、 Stage、 Task 、 Job

RDD(Resilient Distributed Datasets),弹性分布式数据集
DAG(Directed Acyclic Graph),有向无环图


RDD RDD 是 Spark 的灵魂&＃xff0c;也称为弹性分布式数据集。一个 RDD 代表一个可以被分区的只读数据集。RDD 内部可以有许多分区(partitions)&＃xff0c;每个分区又拥有大量的记录(records)。

DAG Spark 中使用 DAG 对 RDD 的关系进行建模&＃xff0c;描述了 RDD 的依赖关系&＃xff0c;这种关系也被称之为 lineage&＃xff08;血缘&＃xff09;&＃xff0c;RDD 的依赖关系使用 Dependency 维护。

Stage 在 DAG 中又进行 Stage 的划分&＃xff0c;划分的依据是依赖是否是 shuffle 的&＃xff0c;每个 Stage 又可以划分成若干 Task。接下来的事情就是 Driver 发送 Task 到 Executor&＃xff0c;Executor 线程池去执行这些 task&＃xff0c;完成之后将结果返回给 Driver。

Job Spark 的 Job 来源于用户执行 action 操作&＃xff08;这是 Spark 中实际意义的 Job&＃xff09;&＃xff0c;就是从 RDD 中获取结果的操作&＃xff0c;而不是将一个 RDD 转换成另一个 RDD 的 transformation 操作。

Task 一个 Stage 内&＃xff0c;最终的 RDD 有多少个 partition&＃xff0c;就会产生多少个 task。

Spark 作业提交流程

1. spark-submit 提交代码&＃xff0c;执行 new SparkContext()&＃xff0c;在 SparkContext 里构造 DAGScheduler 和 TaskScheduler。
2. TaskScheduler 会通过后台的一个进程&＃xff0c;连接 Master&＃xff0c;向 Master 注册 Application。
3. Master 接收到 Application 请求后&＃xff0c;会使用相应的资源调度算法&＃xff0c;在 Worker 上为这个 Application 启动多个 Executor
4. Executor 启动后&＃xff0c;会自己反向注册到 TaskScheduler 中。所有 Executor 都注册到 Driver 上之后&＃xff0c;SparkContext 结束初始化&＃xff0c;接下来往下执行我们自己的代码。
5. 每执行到一个 Action&＃xff0c;就会创建一个 Job。Job 会提交给 DAGScheduler。
6. DAGScheduler 会将 Job 划分为多个 Stage&＃xff0c;然后每个 Stage 创建一个 TaskSet。
7. TaskScheduler 会把每一个 TaskSet 里的 Task&＃xff0c;提交到 Executor 上执行。
8. Executor 上有线程池&＃xff0c;每接收到一个 Task&＃xff0c;就用 TaskRunner 封装&＃xff0c;然后从线程池里取出一个线程执行这个 task。(TaskRunner 将我们编写的代码&＃xff0c;拷贝&＃xff0c;反序列化&＃xff0c;执行 Task&＃xff0c;每个 Task 执行 RDD 里的一个 partition)

Spark 的 Local 和 Standalone

Spark一共有6种运行模式&＃xff1a;Local&＃xff0c;Standalone&＃xff0c;Yarn-Cluster&＃xff0c;Yarn-Client, Mesos, Kubernetes

1. Local: Local 模式即单机模式&＃xff0c;如果在命令语句中不加任何配置&＃xff0c;则默认是 Local 模式&＃xff0c;在本地运行。这也是部署、设置最简单的一种模式&＃xff0c;所有的 Spark 进程都运行在一台机器或一个虚拟机上面。
2. Standalone: Standalone 是 Spark 自身实现的资源调度框架。如果我们只使用 Spark 进行大数据计算&＃xff0c;不使用其他的计算框架时&＃xff0c;就采用 Standalone 模式就够了&＃xff0c;尤其是单用户的情况下。Standalone 模式是 Spark 实现的资源调度框架&＃xff0c;其主要的节点有 Client 节点、Master 节点和 Worker 节点。其中 Driver 既可以运行在 Master 节点上中&＃xff0c;也可以运行在本地 Client 端。当用 spark-shell 交互式工具提交 Spark 的 Job 时&＃xff0c;Driver 在 Master 节点上运行&＃xff1b;当使用 spark-submit 工具提交 Job 或者在 Eclipse、IDEA 等开发平台上使用 new SparkConf.setManager(“spark://master:7077”) 方式运行 Spark 任务时&＃xff0c;Driver 是运行在本地 Client 端上的。

Standalone 模式的部署比较繁琐&＃xff0c;不过官方有提供部署脚本&＃xff0c;需要把 Spark 的部署包安装到每一台节点机器上&＃xff0c;并且部署的目录也必须相同&＃xff0c;而且需要 Master 节点和其他节点实现 SSH 无密码登录。启动时&＃xff0c;需要先启动 Spark 的 Master 和 Slave 节点。提交命令类似于:

./bin/spark-submit \--class org.apache.spark.examples.SparkPi \--master spark://Oscar-2.local:7077 \/tmp/spark-2.2.0-bin-hadoop2.7/examples/jars/spark-examples_2.11-2.2.0.jar \100


其中 master:7077是 Spark 的 Master 节点的主机名和端口号&＃xff0c;当然集群是需要提前启动。

不管使用什么模式&＃xff0c;Spark应用程序的代码是一模一样的&＃xff0c;只需要在提交的时候通过–master参数来指定我们的运行模式即可

• Client
  Driver运行在Client端(提交Spark作业的机器)
  Client会和请求到的Container进行通信来完成作业的调度和执行&＃xff0c;Client是不能退出的
  日志信息会在控制台输出&＃xff1a;便于我们测试

• Cluster
  Driver运行在ApplicationMaster中
  Client只要提交完作业之后就可以关掉&＃xff0c;因为作业已经在YARN上运行了
  日志是在终端看不到的&＃xff0c;因为日志是在Driver上&＃xff0c;只能通过yarn logs -applicationIdapplication_id

./bin/spark-submit \
--class org.apache.spark.examples.SparkPi \
--master yarn \
--executor-memory 1G \
--num-executors 1 \
/home/hadoop/app/spark-2.1.0-bin-2.6.0-cdh5.7.0/examples/jars/spark-examples_2.11-2.1.0.jar \

此处的yarn就是我们的yarn client模式
如果是yarn cluster模式的话&＃xff0c;yarn-cluster

Exception in thread "main" java.lang.Exception: When running with master &＃39;yarn&＃39; either HADOOP_CONF_DIR or YARN_CONF_DIR must be set in the environment.

如果想运行在YARN之上&＃xff0c;那么就必须要设置HADOOP_CONF_DIR或者是YARN_CONF_DIR

1&＃xff09; export HADOOP_CONF_DIR&＃61;/home/hadoop/app/hadoop-2.6.0-cdh5.7.0/etc/hadoop
2) $SPARK_HOME/conf/spark-env.sh

./bin/spark-submit \
--class org.apache.spark.examples.SparkPi \
--master yarn-cluster \
--executor-memory 1G \
--num-executors 1 \
/home/hadoop/app/spark-2.1.0-bin-2.6.0-cdh5.7.0/examples/jars/spark-examples_2.11-2.1.0.jar \
4


yarn logs -applicationId application_1495632775836_0002

宽依赖、窄依赖

1. 窄依赖指的是每一个 Parent RDD 的 Partition 最多被子 RDD 的一个 Partition 使用&＃xff08;一子一亲&＃xff09;

2. 宽依赖指的是多个子 RDD 的 Partition 会依赖同一个 parent RDD的 partition&＃xff08;多子一亲&＃xff09;

RDD 作为数据结构&＃xff0c;本质上是一个只读的分区记录集合。一个 RDD 可以包含多个分区&＃xff0c;每个分区就是一个数据集片段。

首先&＃xff0c;窄依赖可以支持在同一个节点上&＃xff0c;以 pipeline 形式执行多条命令&＃xff08;也叫同一个 Stage 的操作&＃xff09;&＃xff0c;例如在执行了 map 后&＃xff0c;紧接着执行 filter。相反&＃xff0c;宽依赖需要所有的父分区都是可用的&＃xff0c;可能还需要调用类似 MapReduce 之类的操作进行跨节点传递。

其次&＃xff0c;则是从失败恢复的角度考虑。窄依赖的失败恢复更有效&＃xff0c;因为它只需要重新计算丢失的 parent partition 即可&＃xff0c;而且可以并行地在不同节点进行重计算&＃xff08;一台机器太慢就会重新调度到多个节点进行&＃xff09;。

Spark SQL比 Hive 快在哪

当Map的输出结果要被Reduce使用时&＃xff0c;输出结果需要按key哈希&＃xff0c;并且分发到每一个Reducer上去&＃xff0c;这个过程就是shuffle。
由于shuffle涉及到了磁盘的读写和网络的传输&＃xff0c;因此shuffle性能的高低直接影响到了整个程序的运行效率。

Spark SQL 比 Hadoop Hive 快&＃xff0c;是有一定条件的&＃xff0c;而且不是 Spark SQL 的引擎比 Hive 的引擎快&＃xff0c;相反&＃xff0c;Hive 的 HQL 引擎还比 Spark SQL 的引擎更快。其实&＃xff0c;关键还是在于 Spark 本身快。

1. 消除了冗余的 HDFS 读写: Hadoop 每次 shuffle 操作后&＃xff0c;必须写到磁盘&＃xff0c;而 Spark 在 shuffle 后不一定落盘&＃xff0c;可以 persist 到内存中&＃xff0c;以便迭代时使用。如果操作复杂&＃xff0c;很多的 shufle 操作&＃xff0c;那么 Hadoop 的读写 IO 时间会大大增加&＃xff0c;也是 Hive 更慢的主要原因了。

2. 消除了冗余的 MapReduce 阶段: Hadoop 的 shuffle 操作一定连着完整的 MapReduce 操作&＃xff0c;冗余繁琐。而 Spark 基于 RDD 提供了丰富的算子操作&＃xff0c;且 reduce 操作产生 shuffle 数据&＃xff0c;可以缓存在内存中 。

3. JVM 的优化: Hadoop 每次 MapReduce 操作&＃xff0c;启动一个 Task 便会启动一次 JVM&＃xff0c;基于进程的操作。而 Spark 每次 MapReduce 操作是基于线程的&＃xff0c;只在启动 Executor 是启动一次 JVM&＃xff0c;内存的 Task 操作是在线程复用的。每次启动 JVM 的时间可能就需要几秒甚至十几秒&＃xff0c;那么当 Task 多了&＃xff0c;这个时间 Hadoop 不知道比 Spark 慢了多少。

打包的注意事项

打包时要注意&＃xff0c;pom.xml中需要添加如下plugin

<plugin><artifactId>maven-assembly-pluginartifactId><configuration><archive><manifest><mainClass>mainClass>manifest>archive><descriptorRefs><descriptorRef>jar-with-dependenciesdescriptorRef>descriptorRefs>configuration>
plugin>


mvn assembly:assembly

./bin/spark-submit \
--class com.hiszm.log.SparkStatCleanJobYARN \
--name SparkStatCleanJobYARN \
--master yarn \
--executor-memory 1G \
--num-executors 1 \
--files /home/hadoop/lib/ipDatabase.csv,/home/hadoop/lib/ipRegion.xlsx \
/home/hadoop/lib/sql-1.0-jar-with-dependencies.jar \
hdfs://hadoop001:8020/hiszm/input/* hdfs://hadoop001:8020/hiszm/clean


注意&＃xff1a;–files在spark中的使用

spark.read.format("parquet").load("/hiszm/clean/day&＃61;20170511/part-00000-71d465d1-7338-4016-8d1a-729504a9f95e.snappy.parquet").show(false)

./bin/spark-submit \
--class com.hiszm.log.TopNStatJobYARN \
--name TopNStatJobYARN \
--master yarn \
--executor-memory 1G \
--num-executors 1 \
/home/hadoop/lib/sql-1.0-jar-with-dependencies.jar \
hdfs://hadoop001:8020/hiszm/clean 20170511


存储格式的选择&＃xff1a;http://www.infoq.com/cn/articles/bigdata-store-choose/
压缩格式的选择&＃xff1a;https://www.ibm.com/developerworks/cn/opensource/os-cn-hadoop-compression-analysis/

调整并行度

./bin/spark-submit \
--class com.hiszm.log.TopNStatJobYARN \
--name TopNStatJobYARN \
--master yarn \
--executor-memory 1G \
--num-executors 1 \
--conf spark.sql.shuffle.partitions&＃61;100 \
/home/hadoop/lib/sql-1.0-jar-with-dependencies.jar \
hdfs://hadoop001:8020/hiszm/clean 20170511