spark sql 在mysql的应用实践

骑猪看日落 2022-05-31 08:48 240阅读 0赞

# 前言 #

目前spark sql 主要应用在structure streaming、etl 和 machine learning 的场景上， 它能对结构化的数据进行存储和操作，结构化的数据可以来自HIve、JSON、Parquet、JDBC/ODBC等数据源。由于部门对数据的准确性，一致性和维护等等要求等业务特点，我们选择mysql使用jdbc的方式作为我们的数据源，spark 集群用yarn做为资源管理，本文主要分享我们在使用spark sql 过程中遇到的问题和一些常用的开发实践总结。

运行环境：spark :2.1.0，hadoop: hadoop-2.5.0-cdh5.3.2 (yarn 资源管理，hdfs)，mysql：5.7 ，scala: 2.11， java：1.8

--------------------

# spark on yarn  #

![spark on yarn 运行机制][spark on yarn]

我们先来了解一下spark on yarn 任务的运行机制。**yarn** 的基本思想是将**JobTracker**的两个主要功能（资源管理和任务调度/监控）分离成单独的组件：RM 和 AM；新的资源管理器**ResourceManager(RM)**实现全局的所有应用的计算资源分配，应用控制器**ApplicationMaster(AM)**实现应用的调度和资源的协调；节点管理器**NodeManager(NM)**则是每台机器的代理，处理来自AM的命令，实现节点的监控与报告；容器 **Container** 封装了内存、CPU、磁盘、网络等资源，是资源隔离的基础，当AM向RM申请资源时，RM为AM返回的资源便是以Container表示，如上图，spark master分配的 executor 的执行环境便是containner。目前我们使用yarn 队列的方式，可以进一步的对应用执行进行管理，让我们的应用分组和任务分配更加清晰和方便管理。

![yarn 队列][yarn]

--------------------

# 开发实践 #

### 1. 读取mysql表数据 ###

import com.test.spark.db.ConnectionInfos;
    import org.apache.spark.sql.Dataset;
    import org.apache.spark.sql.Row;
    import org.apache.spark.sql.SparkSession;
    
    import java.util.Arrays;
    
    public class SparkSimple01 {
    
        public static void main(String[] args) {
    
            // 创建spark会话，实质上是SQLContext和HiveContext的组合
            SparkSession sparkSession = SparkSession.builder().master("local[*]").appName("Java Spark SQL basic example").getOrCreate();
    
            // 设置日志级别，默认会打印DAG,TASK执行日志，设置为WARN之后可以只关注应用相关日志
            sparkSession.sparkContext().setLogLevel("WARN");
    
            // 分区方式读取mysql表数据
            Dataset<Row> predicateSet = sparkSession.read().jdbc(ConnectionInfos.TEST_MYSQL_CONNECTION_URL, "people",
                    (String[]) Arrays.asList(" name = 'tom'", " name = 'sam' ").toArray(), ConnectionInfos.getTestUserAndPasswordProperties());
    
            predicateSet.show();
    
        }
    }

为了确认该查询对mysql发出的具体sql，我们先查看一下mysql执行sql日志，

#mysql 命令窗口执行以下命令打开日志记录
    SHOW VARIABLES LIKE "general_log%";
    SET GLOBAL general_log = 'ON';

![mysql log.png][]

打开Lenovo.log得到以上代码在mysql上的执行情况：  
![分区执行sql][sql]

通过分区查询获取表数据的方式有以下几个优点：  
\+ 利用表索引查询提高查询效率  
\+ 自定义sql条件使分区数据更加均匀，方便后面的并行计算  
\+ 分区并发读取可以通过控制并发控制对mysql的查询压力  
\+ 可以读取大数据量的mysql表

spark jdbc 读取msyql表还有直接读取(无法读取大数据量表)，指定字段分区读取(分区不够均匀)等方式，通过项目实践总结，以上的分区读取方式是我们目前认为对mysql最友好的方式。  
分库分表的系统也可以利用这种方式读取各个表在内存中union所有spark view得到一张统一的内存表，在业务操作中将分库分表透明化。如果线上数据表数据量较大的时候，在union之前就需要将spark view通过**指定字段的方式查询**，避免on line ddl 在做变更时union表报错，因为可能存在部分表已经添加新字段，部分表还未加上新字段，而union要求所有表的表结构一致，导致报错。

### 2. Dataset 分区数据查看 ###

我们都知道 Dataset 的分区是否均匀，对于结果集的并行处理效果有很重要的作用，spark Java版暂时无法查看partition分区中的数据分布，这里用java调用scala 版api方式查看，线上不推荐使用，因为这里的分区查看使用foreachPartition，多了一次action操作，并且打印出全部数据。

import org.apache.spark.sql.{Dataset, Row}
    
    /** * Created by lesly.lai on 2017/12/25. */
    class SparkRddTaskInfo { 
      def getTask(dataSet: Dataset[Row]) {
        val size = dataSet.rdd.partitions.length
        println(s"==> partition size: $size " )
        import scala.collection.Iterator
        val showElements = (it: Iterator[Row]) => {
          val ns = it.toSeq
          import org.apache.spark.TaskContext
          val pid = TaskContext.get.partitionId
          println(s"[partition: $pid][size: ${ns.size}] ${ns.mkString(" ")}")
        }
        dataSet.foreachPartition(showElements)
      }
    }

还是用上面读取mysql数据的例子来演示调用，将predicateSet作为参数传入

new SparkRddTaskInfo().getTask(predicateSet);

控制台打印结果  
![分区结果.png][.png]

通过分区数据，我们可以看到之前的predicate 方式得到的分区数就是predicate size 大小，并且按照我们想要的数据分区方式分布数据，这对于业务数据的批处理，executor的local cache，spark job执行参数调优都很有帮助，例如调整spark.executor.cores，spark.executor.memory，GC方式等等。  
这里涉及java和Scala容器转换的问题，Scala和Java容器库有很多相似点，例如，他们都包含迭代器、可迭代结构、集合、 映射和序列。但是他们有一个重要的区别。Scala的容器库特别强调不可变性，因此提供了大量的新方法将一个容器变换成一个新的容器。  
在Scala内部，这些转换是通过一系列“包装”对象完成的，这些对象会将相应的方法调用转发至底层的容器对象。所以容器不会在Java和Scala之间拷贝来拷贝去。一个值得注意的特性是，如果你将一个Java容器转换成其对应的Scala容器，然后再将其转换回同样的Java容器，最终得到的是一个和一开始完全相同的容器对象（这里的相同意味着这两个对象实际上是指向同一片内存区域的引用，容器转换过程中没有任何的拷贝发生）。

### 3. sql 自定义函数 ###

自定义函数，可以简单方便的实现业务逻辑。

import com.tes.spark.db.ConnectionInfos;
    import org.apache.spark.sql.Dataset;
    import org.apache.spark.sql.Row;
    import org.apache.spark.sql.SparkSession;
    import org.apache.spark.sql.types.DataTypes;
    
    public class SparkSimple02 {
        public static void main(String[] args) {
            SparkSession sparkSession = SparkSession.builder().master("local[*]").appName("Java Spark SQL basic example").getOrCreate();
            sparkSession.sparkContext().setLogLevel("WARN");
            Dataset<Row> originSet = sparkSession.read().jdbc(ConnectionInfos.TEST_MYSQL_CONNECTION_URL, "people", ConnectionInfos.getTestUserAndPasswordProperties());
            originSet.cache().createOrReplaceTempView("people");
    
            // action操作 打印原始结果集
            originSet.show();
    
            // 注册自定义函数
            sparkSession.sqlContext().udf().register("genderUdf", gender -> {
                if("M".equals(gender)){
                    return  "男";
                }else if("F".equals(gender)){
                    return  "女";
                }
                return "未知";
            }, DataTypes.StringType);
    
            // 查询结果
            Dataset<Row> peopleDs = sparkSession.sql("select job_number,name,age, genderUdf(gender) gender, dept_id, salary, create_time from people ");
    
            // action操作 打印函数处理后结果集
            peopleDs.show();
        }
    }

执行结果：  
![image.png][]

在sql中用使用java代码实现逻辑操作，这为sql的处理逻辑能力提升了好几个层次，将函数抽取成接口实现类可以方便的管理和维护这类自定义函数类。此外，spark也支持自定义内聚函数，窗口函数等等方式，相比传统开发实现的功能方式，使用spark sql开发效率可以明显提高。

### 4. mysql 查询连接复用 ###

最近线上任务遇到一个获取mysql connection blocked的问题，从spark ui的executor thread dump 可以看到blocked的栈信息，如图：  
![connection blocked.png][]

查看代码发现DBConnectionManager 调用了 spark driver注册mysql driver 使用同步方式的代码  
![driverRegister.png][]  
看到这里我们很容易觉得是注册driver 导致的blocked，其实再仔细看回报错栈信息，我们会发现，这里的getConnection是在dataset 的foreachpartition 中调用，并且是在每次db 操作时获取一次getConnection 操作，这意味着在该分区下有多次重复的在同步方法中注册driver获取连接的操作，看到这里线程blocked的原因就很明显了，这里我们的解决方式是：  
a. 在同个partition中的connection 复用进行db操作  
b. 为了避免partition数据分布不均导致连接active时间过长，加上定时释放连接再从连接池重新获取连接操作  
通过以上的连接处理，解决了blocked问题，tps也达到了4w左右。

### 5. executor 并发控制 ###

我们都知道，利用spark 集群分区并行能力，可以很容易实现较高的并发处理能力，如果是并发的批处理，那并行处理的能力可以更好，但是，mysql 在面对这么高的并发的时候，是有点吃不消的，因此我们需要适当降低spark 应用的并发和上下游系统和平相处。控制spark job并发可以通过很多参数配置组合、集群资源、yarn队列限制等方式实现，经过实践，我们选择以下参数实现：

#需要关闭动态内存分配，其他配置才生效
    spark.dynamicAllocation.enabled = false
    spark.executor.instances = 2
    spark.executor.cores = 2

![image.png][image.png 1]

这里发现除了设置executor配置之外，还需要关闭spark的动态executor分配机制，spark 的ExecutorAllocationManager 是 一个根据工作负载动态分配和删除 executors 的管家， ExecutorAllocationManager 维持一个动态调整的目标executors数目， 并且定期同步到资源管理者，也就是 yarn ，启动的时候根据配置设置一个目标executors数目， spark 运行过程中会根据等待（pending）和正在运行(running)的tasks数目动态调整目标executors数目，因此需要关闭动态配置资源才能达到控制并发的效果。

除了executor是动态分配之外，Spark 1.6 之后引入的统一内存管理机制，与静态内存管理的区别在于存储内存和执行内存共享同一块空间，可以动态占用对方的空闲区域，我们先看看worker中的内存规划是怎样的：  
![worker memory schedule.png][]

worker 可以根据实例配置，内存配置，cores配置动态生成executor数量，每一个executor为一个jvm进程，因此executor 的内存管理是建立在jvm的内存管理之上的。从本文第一张spark on yarn图片可以看到，yarn模式的 executor 是在yarn container 中运行，因此container的内存分配大小同样可以控制executor的数量。  
RDD 的每个 Partition 经过处理后唯一对应一个 Block（BlockId 的格式为 rdd\_RDD-ID\_PARTITION-ID ），从上图可以看出，开发过程中常用的分区（partition）数据是以block的方式存储在堆内的storage内存区域的，还有为了减少网络io而做的broadcast数据也存储在storage区域；堆内的另一个区域内存则主要用于缓存rdd shuffle产生的中间数据；此外，worker 中的多个executor还共享同一个节点上的堆外内存，这部分内存主要存储经序列化后的二进制数据，使用的是系统的内存，可以减少不必要的开销以及频繁的GC扫描和回收。

为了更好的理解executor的内存分配，我们再来看一下executor各个内存块的参数设置：  
![executor jvm][]  
![off-heap.png][]

了解spark 内存管理的机制后，就可以根据mysql的处理能力来设置executor的并发处理能力，让我们的spark 应用处理能力收放自如。调整executor数量还有另外一个好处，就是集群资源规划，目前我们的集群队列是yarn fair 模式，  
![yarn fair 集群模式.png][yarn fair _.png]

先看看yarn fair模式，举个例子，假设有两个用户A和B，他们分别拥有一个队列。当A启动一个job而B没有任务时，A会获得全部集群资源；当B启动一个job后，A的job会继续运行，当A的job执行完释放资源后，不过一会儿之后两个任务会各自获得一半的集群资源。如果此时B再启动第二个job并且其它job还在运行，则它将会和B的第一个job共享B这个队列的资源，也就是B的两个job会用于四分之一的集群资源，而A的job仍然用于集群一半的资源，结果就是资源最终在两个用户之间平等的共享。

在这种情况下，即使有多个队列执行任务，fair模式容易在资源空闲时占用其他队列资源，一旦占用时间过长，就会导致其他任务都卡住，这也是我们遇到的实际问题。如果我们在一开始能评估任务所用的资源，就可以在yarn队列的基础上指定应用的资源，例如executor的内存，cpu，实例个数，并行task数量等等参数来管理集群资源，这有点类似于yarn Capacity Scheduler 队列模式，但又比它有优势，因为spark 应用可以通过spark context的配置来动态的设置，不用在配置yarn 队列后重启集群，稍微灵活了一点。

除了以上提到的几点总结，我们还遇到很多其他的疑问和实践，例如，什么时候出现shuffle；如何比较好避开或者利用shuffle；Dataset 的cache操作会不会有性能问题，如何从spark ui中分析定位问题；spark 任务异常处理等等，暂时到这里，待续…

赖泽坤 @vip.fcs

参考资料：  
[http://www.cnblogs.com/yangsy0915/p/5118100.html][http_www.cnblogs.com_yangsy0915_p_5118100.html]  
[https://mp.weixin.qq.com/s/KhHy1mURJBiPMGqkl4-JEw][https_mp.weixin.qq.com_s_KhHy1mURJBiPMGqkl4-JEw]  
[https://www.ibm.com/developerworks/cn/analytics/library/ba-cn-apache-spark-memory-management/index.html?ca=drs-&utm\_source=tuicool&utm\_medium=referral][https_www.ibm.com_developerworks_cn_analytics_library_ba-cn-apache-spark-memory-management_index.html_ca_drs-_utm_source_tuicool_utm_medium_referral]  
[https://docs.scala-lang.org/zh-cn/overviews/collections/conversions-between-java-and-scala-collections.html][https_docs.scala-lang.org_zh-cn_overviews_collections_conversions-between-java-and-scala-collections.html]  
[https://www.jianshu.com/p/e7db5970e68c][https_www.jianshu.com_p_e7db5970e68c]

[spark on yarn]: /images/20220531/c983681f379247ac89132892838b686b.png
[yarn]: /images/20220531/ecdbd480470349c09c3067420df984c8.png
[mysql log.png]: /images/20220531/3fe028fc99dd42c6b32f44c8df5d101e.png
[sql]: /images/20220531/43f973f730a14f099062e4c3ec790dd6.png
[.png]: /images/20220531/5ca2d428a13b43c0a3b5a3038e11a8ea.png
[image.png]: /images/20220531/3ffd027bbfca4ab7b802411f93d84627.png
[connection blocked.png]: /images/20220531/f822f616fdd547df8db74eb402b27d76.png
[driverRegister.png]: /images/20220531/a361c43e3d4840ef973995bbc95dc399.png
[image.png 1]: /images/20220531/b6fa06a141da4f81a1ea8248f50ceb21.png
[worker memory schedule.png]: /images/20220531/36fe082fd88e4941b98376441fad5f0a.png
[executor jvm]: /images/20220531/6c18a52d62124361861cf0396acd868a.png
[off-heap.png]: /images/20220531/05d4c8ae0aae4374b273fd610e22aef3.png
[yarn fair _.png]: /images/20220531/b12004f3b51a4905bc4b7ab272062043.png
[http_www.cnblogs.com_yangsy0915_p_5118100.html]: http://www.cnblogs.com/yangsy0915/p/5118100.html
[https_mp.weixin.qq.com_s_KhHy1mURJBiPMGqkl4-JEw]: https://mp.weixin.qq.com/s/KhHy1mURJBiPMGqkl4-JEw
[https_www.ibm.com_developerworks_cn_analytics_library_ba-cn-apache-spark-memory-management_index.html_ca_drs-_utm_source_tuicool_utm_medium_referral]: https://www.ibm.com/developerworks/cn/analytics/library/ba-cn-apache-spark-memory-management/index.html?ca=drs-&utm_source=tuicool&utm_medium=referral
[https_docs.scala-lang.org_zh-cn_overviews_collections_conversions-between-java-and-scala-collections.html]: https://docs.scala-lang.org/zh-cn/overviews/collections/conversions-between-java-and-scala-collections.html
[https_www.jianshu.com_p_e7db5970e68c]: https://www.jianshu.com/p/e7db5970e68c