Hive SQL迁移Spark SQL在滴滴的实践

桔妹导读： 在滴滴S QL任务从Hive迁移到Spark后，Spark SQL任务占比提升至85%，任务运行时间节省40% ，运行任务需要的计算资源节省21%，内存资源节省49%。在迁移过程 中 我们沉淀出一套迁移流程， 并且发现并解决了两个引擎在语法，UDF，性能和功能方面的差异。

1. 

迁移背景     

Spark自从2010年面世，到2020年已经经过十年的发展，现在已经发展为大数据批计算的首选引擎，在滴滴Spark是在2015年便开始落地使用，不过主要使用的场景是更多在数据挖掘和机器学习方向，对于数仓SQL方向，主要仍以Hive SQL为主。

下图是当前滴滴内部SQL任务的架构图，滴滴各个业务线的离线任务是通过一站式数据开发平台DataStudio调度的，DataStudio把SQL任务提交到HiveServer2或者Spark两种计算引擎上。两个计算引擎均依赖资源管理器YARN和文件系统HDFS。

在迁移之前我们面临的主要问题有：

• SQL任务运行慢： 迁移前SQL任务运行的平均时间是20分钟，主要原因是占比高达83%的Hive SQL任务运行时间长，Hive任务执行过程中会启动多个MR Job，Job间的中间结果存储在HDFS，所以同一个SQL， Hive比Spark执行的时间更长；

• Hive SQL稳定性差： 一个HS2会同时执行多个用户的Hive SQL任务，当一个异常任务导致HS2进程响应慢甚至异常退出时，运行在同一个实例的SQL任务也会运行缓慢甚至失败。而异常任务场景各异。我们曾经遇到的异常任务有多个大SQL加载过多的分区元数据导致HS2 FullGC，加载UDF时导致HS2进程core dump，UDF访问HDFS没有关闭流导致HS2机器端口被打满，这些没有通用解法， 问题很难收敛；

• 人力分散： 两个引擎需要投入双倍的人力，在人员有限的情况下，对引擎的掌控力会减弱；

所以为了SQL任务运行 更快，更稳 ，团队人力聚焦，对引擎有 更强 的掌控力，我们决定把Hive SQL迁移到Spark SQL。

2. 

迁移方案概要设计

3. 

迁移方案详细设计

▍ 1. Hive SQL提取

Hive SQL提取包括以下步骤：

• 改造HiveHistoryImpl，每个session内执行的所有SQL和command保存到HiveServer2的一个本地文件中，这些文件按天组织，每天一个目录

• 定时将前一天的History目录上传到hdfs

• 开发HiveHistoryParser

HiveHistoryParser的主要功能是：

• 每天从HDFS下载所有HiveServer2的History文件；

• SQL去重：DataStudio上的一个SQL任务可能一天执行多次(比如小时任务)，任务执行一次会生成一个新的执行Id，只保留一天中最大的执行Id的SQL；

• 合并SQL：一个shell任务可能建立多个session执行SQL，为了后面迁移shell任务，需要把多个session的SQL合并到一起；

• 输出Parse结果：包括多个SQL文件和meta文件:

  • 每个任务执行的SQL保存到一个文件中，文件名是任务名称加执行Id，我们称作 原始SQL文件 ；

  • meta文件包含SQL文件路径，任务名称，项目名称，用户名；

▍ 2. SQL改写&双跑

SQL改写会对上一步生成的每个原始SQL文件执行以下步骤：

• 使用Spark的SessionState对SQL文件逐行分析，识别是否包含以下两类子句：

  • insert overwrite into

  • create table as select

• 如果包含上面的两类子句，则提取写入的目标库表名称；

• 在测试库中创建与目标库表schema完全一致的两个测试表；

• 分别使用上一步创建的测试库表替换原始SQL文件中的库表名生成用于回放的SQL文件，一个原始SQL文件改写后会生成两个SQL文件，用于后面两个引擎分别执行；

SQL双跑步骤如下：

• 并发的使用Spark和Hive执行上一步生成的两个SQL文件；

• 记录使用两种引擎执行SQL时启动的Application和运行时间；

• 输出回放结果到文件中，执行每个SQL文件对会生成一条结果记录， 包括Hive 和Spark 执行SQL的时间，启动的Application列表，和输出的目标库表名称等， 如下图所示：

▍ 3. 结果对比

结果对比时会遍历每个回放记录，统计以下指标：

具体流程如下：

• 查询Spark SQL和Hive SQL输出的库表的记录数；

• 查询两种引擎输出的HDFS文件个数和大小；

• 对比两种引擎的输出数据；

分别对Spark和Hive的产出表执行以下SQL，获取表的概要信息

比较两张表的概要信息：

• 如果所有对应列的值相同则认为结果一致；

• 如果存在不一致的列，如果该列是数值类型，则对该列计算最大精度差异， SQL如下：

• 统计两种引擎启动的Application消耗的vcore和memory资源；

• 输出对比结果， 包括运行时间， 消耗的vcore和memory，是否一致，如果不一致输出不一致的列名以及最大差异；

• 汇总数据结果，并对回放的SQL分为以下几类：

• • 可迁移： 数据完全一致， 并且使用Spark SQL执行使用更少资源，包括运行时间，vcore和memory以及文件数；

  • 经验可迁移： 在排查不一致时发现有些是逻辑正确的 (比如collect_set结果顺序不一致)，如果有些任务符合这些经验，则认为是经验可迁移；

  • 数据不一致： 两种引擎产出的结果存在不一致的列，而且没有命中经验；

  • Time_High： 两种引擎产出的结果完全一致，但是Spark执行SQL的运行时间大于Hive执行SQL的时间；

  • Cpu_High： 两种引擎产出的结果完全一致，但是Spark执行SQL消耗的cpu资源大于Hive执行SQL消耗的cpu资源；

  • Memory_High： 两种引擎产出的结果完全一致，但是Spark执行SQL消耗的memory资源大于Hive执行SQL消耗的memory资源；

  • Files_High： 两种引擎产出的结果完全一致，但是Spark执行SQL产生的文件数大于Hive执行SQL产生的文件数；

  • 语法不兼容： 在SQL改写阶段解析SQL时报语法错误；

  • 运行时异常： 在双跑阶段，Hive SQL或者Spark SQL在运行过程中失败；

▍ 4. 迁移

迁移比较简单， 步骤如下：

• 整理迁移任务列表以及对应的配置参数；

• 调用DataStudio接口把任务类型修改为SparkSQL类型；

• 重跑任务；

▍ 5. 问题排查&修复

如果SQL是“可迁移”或者“经验可迁移”，可以执行迁移，其它的任务需要排查，这部分是最耗时耗力的，迁移过程中大部分时间都是在调查和修复这些问题。修复之后再执行从头开始，提取最新任务的SQL，然后SQL改写和双跑，结果对比，满足迁移条件则说明修复了问题，可以迁移，否则继续排查，因此迁移过程是一个循环往复的过程，直到SQL满足迁移条件，整体过程如下图所示：

4. 

引擎差异

在迁移的过程中我们发现了很多两种引擎不同的地方，主要包括语法差异，UDF差异，功能差异和性能差异。

▍ 1. 语法差异

有些Hive SQL使用Spark SQL执行在语法分析阶段就会出错，有些语法差异我们在内部版本已经修复，目前正在反馈社区，正在和社区讨论，还有一些目前没有修复。

1.1  用例设计

• UDTF新版initialize接口支持，对齐Hive SQL  [SPARK-33704]

• Window Function 不支持没有order by子句的场景

• Join 子查询支持rand 随机分布条件，增强语法兼容

• Orc/Orcfile 存储类型创建语句屏蔽ROW FORMAT DELIMITED限制 [SPARK-33755]

• `DB.TB` 识别支持，对齐Hive SQL [SPARK-33686]

• 支持CREATE TEMPORARY TABLE

• 各类Hive UDF的支持调用，主要包括get_json_object，datediff，unix_timestamp，to_date，collect_set，date_sub  [SPARK-33721]

• DROP不存在的表和分区，Spark SQL报错，Hive SQL 正常  [SPARK-33637]

• 删除分区时支持设置过滤条件 [SPARK-33691]

1.2  未修复

• Map类型字段不支持GROUP BY操作

• Operation not allowed：ALTER TABLE CONCATENATE

▍ 2.  UDF差异

在排查数据不一致的SQL过程中，我们发现有些是因为输入数据的顺序不同造成的， 这些差异逻辑上是正确的，而有些是UDF对异常值的处理方式不一致造成的，还有需要注意的是UDF执行环境不同造成的结果差异。

2.1 顺序差异

这些因为输入数据的顺序不同造成的结果差异逻辑上是一致的，对业务无影响，因此在迁移过程中可以忽略这些差异，这类差异的SQL任务属于经验可迁移。

2 .1.1  collect_set

• 假设数据表如下：

• 执行如下SQL：

• 执行结果：

• 差异说明：

collect_set执行结果的顺序取决于记录被扫描的顺序，Spark SQL执行过程中是多个任务并发执行的，因此记录被读取的顺序是无法保证的.

2 .1.2  collect_list

• 假设数据表如下：

• 执行如下SQL：

• 执行结果：

• 差异说明：

• 执行如下SQL：

• 执行结果：

• 差异说明：

Map类型是无序的，同一份数据，在query时显示的各个key的顺序会有变化。

2 .1.5  sum(double/float)

• 假设数据表如下：

• 执行如下SQL：

• 执行结果：

• 差异说明：

这是由float/double类型的表示方式决定的，浮点数不能表示所有的实数，在执行运算过程中会有精度丢失，对于几个浮点数，执行加法时的顺序不同，结果有时就会不同。

2 .1.6  顺序差异解决方案

由以上UDF造成的差异可以忽略，相关任务如果在资源方面也有节省，那么最终的状态是经验可迁移状态，符合迁移条件。

2.2  非顺序差异

下面几个日期/时间相关函数，当有异常输入是Spark SQL会返回NULL，而Hive SQL会返回一个非NULL值。

2 .2.1  datediff

对于异常日期，比如0000-00-00执行datediff两者会存在差异。

2 .2.2  unix_timestamp

对于24点Spark认为是非法的返回NULL，而Hive任务是正常的，下表时执行unix_timestamp(concat('2020-06-01', ' 24:00:00'))时的差异。

2 .2.3  to_date

当月或者日是00时Hive仍然会返回一个日期，但是Spark会返回NULL。

2 .2.4 date_sub

当月或者日是00时Hive仍然会返回一个日期，但是Spark会返回NULL。

2 .2.5  date_add

当月或者日是00时Hive仍然会返回一个日期，但是Spark会返回NULL。

2 .2.6  非顺序差异解决方案

这些差异是是因为对异常UDF参数的处理逻辑不同造成的，虽然Spark SQL返回NULL更合理，但是现有的Hive SQL任务用户适应了这种处理逻辑，所以为了不影响现有SQL任务，我们对这类UDF做了兼容处理，用户可以通过配置来决定使用Hive内置函数还是Spark的内置UDF。

2.3 UDF执行环境差异

2 .3.1  差异说明

基于MapReduce的Hive SQL一个Task会启动一个进程，进程中的主线程负责数据处理， 因此在Hive SQL中UDF只会在单程中执行。

而Spark 一个Executor可能会启动多个Task，如下图所示。因此在Spark SQL中自定义UDF时需要考虑线程安全问题。

2 .3.2 差异解决方案

下面是一个非线程安全的示例，UDF内部共享静态变量，在执行UDF时会读写这个静态变量。

解决方案也比较简单，一种是加锁，如下图所示：

5. 

迁移结果

经过6个多月的团队的努力，我们迁移了1万多个Hive SQL任务到Spark SQL，在迁移过程中，随着spark SQL任务的增加，SQL任务的执行时间在逐渐减少，从最初的1000+秒下降到600+秒如下图所示：

迁移后Spark SQL任务占比85%，SQL任务运行时间节省40%，计算资源节省21%，内存资源节省49%，迁移的收益是非常大的。

6. 

下一步计划

迁移之后Spark已经成为SQL任务的主流引擎，但是还有大量的shell类型任务使用Hive执行SQL，所以后续我们会迁移shell类型任务，把shell中的Hive SQL迁移到Spark SQL。

在生产环境中，有些shuffle 比较中的任务经常会因为shuffle fetch重试甚至失败，我们想优化Spark External Shuffle Service。

社区推出Spark 3.x也半年多了，在功能和性能上有很大提升，所以我们也想和社区保持同步，升级Spark到3.x版本。

本文作者

▬

延伸阅读

▬

内容编辑 | Mango

联系我们 | [email protected]