hive取消打印日志信息_舍弃Hive、拥抱SparkSQL，有赞的大数据实践

本文由 「AI前线」原创&＃xff08;ID&＃xff1a;ai-front&＃xff09;&＃xff0c;原文链接&＃xff1a;舍弃Hive、拥抱SparkSQL&＃xff0c;有赞的大数据实践

作者 | 邹晨俊
来源 | 授权转载自微信公众号有赞Coder&＃xff08;ID&＃xff1a;youzan_coder&＃xff09;
编辑 | Natalie

AI 前线导读&＃xff1a;有赞数据平台从 2017 年上半年开始&＃xff0c;逐步使用 SparkSQL 替代 Hive 执行离线任务&＃xff0c;目前 SparkSQL 每天的运行作业数量 5000 个&＃xff0c;占离线作业数目的 55%&＃xff0c;消耗的 CPU 资源占集群总资源的 50% 左右。本文介绍由 SparkSQL 替换 Hive 过程中碰到的问题以及处理经验和优化建议。

更多优质内容请关注微信公众号“AI 前线”&＃xff08;ID&＃xff1a;ai-front&＃xff09;

本文将包括以下内容&＃xff1a;

• 有赞数据平台的整体架构&＃xff1b;
• SparkSQL 在有赞的技术演进&＃xff1b;
• 从 Hive 到 SparkSQL 的迁移之路。

一. 有赞数据平台介绍

首先介绍一下有赞大数据平台总体架构&＃xff1a;

如下图所示&＃xff0c;底层是数据导入部分&＃xff0c;其中 DataY 区别于开源届的全量导入导出工具 alibaba/DataX&＃xff0c;是有赞内部研发的离线 Mysql 增量导入 Hive 的工具&＃xff0c;把 Hive 中历史数据和当天增量部分做合并。DataX / DataY 负责将 Mysql 中的数据同步到数仓当中&＃xff0c;Flume 作为日志数据的主要通道&＃xff0c;同时也是 Mysql binlog 同步到 HDFS 的管道&＃xff0c;供 DataY 做增量合并使用。

第二层是大数据的计算框架&＃xff0c;主要分成两部分&＃xff1a;分布式存储计算和实时计算&＃xff0c;实时框架目前主要支持 JStorm&＃xff0c;Spark Streaming 和 Flink&＃xff0c;其中 Flink 是今年开始支持的&＃xff1b;而分布式存储和计算框架这边&＃xff0c;底层是 Hadoop 和 Hbase&＃xff0c;ETL 主要使用 Hive 和 Spark&＃xff0c;交互查询则会使用 Spark&＃xff0c;Presto&＃xff0c;实时 OLAP 系统今年引入了 Druid&＃xff0c;提供日志的聚合查询能力。

第三层是数据平台部分&＃xff0c;数据平台是直接面对数据开发者的&＃xff0c;包括几部分的功能&＃xff0c;数据开发平台&＃xff0c;包括日常使用的调度&＃xff0c;数据传输&＃xff0c;数据质量系统&＃xff1b;数据查询平台&＃xff0c;包括 ad-hoc 查询以及元数据查询。有关有赞数据平台的详细介绍可以参考往期有赞数据平台的 博客内容。

二. SparkSQL 技术演进

从 2017 年二季度&＃xff0c;有赞数据组的同学们开始了 SparkSQL 方面的尝试&＃xff0c;主要的出发点是当时集群资源是瓶颈&＃xff0c;Hive 跑任务已经逐渐开始乏力&＃xff0c;有些复杂的 SQL&＃xff0c;通过 SQL 的逻辑优化达到极限&＃xff0c;仍然需要几个小时的时间。业务数据量正在不断增大&＃xff0c;这些任务会影响业务对外服务的承诺。同时&＃xff0c;随着 Spark 以及其社区的不断发展&＃xff0c;Spark 及 Spark SQL 本身技术的不断成熟&＃xff0c;Spark 在技术架构和性能上都展示出 Hive 无法比拟的优势。

从开始上线提供离线任务服务&＃xff0c;再到 Hive 任务逐渐往 SparkSQL 迁移&＃xff0c;踩过不少坑&＃xff0c;也填了不少坑&＃xff0c;这里主要分两个方面介绍&＃xff0c;一方面是我们对 SparkSQL 可用性方面的改造以及优化&＃xff0c;另一方面是 Hive 迁移时遇到的种种问题以及对策。

2.1 可用性改造

可用性问题包括两方面&＃xff0c;一个是系统的稳定性&＃xff0c;监控 / 审计 / 权限等&＃xff0c;另一个是用户使用的体验&＃xff0c;用户以前习惯用 Hive&＃xff0c;如果 SparkSQL 的日志或者 Spark thrift server 的 UI 不能够帮助用户定位问题&＃xff0c;解决问题&＃xff0c;那也会影响用户的使用或者迁移意愿。所以我首先谈一下用户交互的问题。

用户体验

我们碰到的第一个问题是用户向我们抱怨通过 JDBC 的方式和 Spark thrift server(STS) 交互&＃xff0c;执行一个 SQL 时&＃xff0c;没有执行的进度信息&＃xff0c;需要一直等待执行成功&＃xff0c;或者任务出错时接收任务报错邮件得知执行完。于是执行进度让用户可感知是一个必要的功能。我们做了 Spark 的改造&＃xff0c;增加运行时的 operation 日志&＃xff0c;并且向社区提交了 patch(spark-22496)&＃xff0c; 而在我们内部&＃xff0c;更增加了执行进度日志&＃xff0c;每隔 2 秒打印出当前执行的 job/stage 的进度&＃xff0c;如下图所示。

监控

SparkSQL 需要收集 STS 上执行的 SQL 的审计信息&＃xff0c;包括提交者执行的具体 SQL&＃xff0c;开始结束时间&＃xff0c;执行完成状态。原生 STS 会把这些信息通过事件的方式 post 到事件总线&＃xff0c;监听者角色 (HiveThriftServer2Listener) 在事件总线上注册&＃xff0c;订阅消费事件&＃xff0c;但是这个监听者只负责 Spark UI 的 JDBC Tab 上的展示&＃xff0c;我们改造了 SparkListener 类&＃xff0c;将 session 以及执行的 sql statement 级别的消息也放到了总线上&＃xff0c;监听者可以在总线上注册&＃xff0c;以便消费这些审计信息&＃xff0c;并且增加了一些我们感兴趣的维度&＃xff0c;如使用的 cpu 资源&＃xff0c;归属的工作流 (airflowId)。同时&＃xff0c;我们增加了一种新的完成状态 cancelled&＃xff0c;以方便区分是用户主动取消的任务。

Thrift Server HA

相比于 HiveServer&＃xff0c;STS 是比较脆弱的&＃xff0c;一是由于 Spark 的 driver 是比较重的&＃xff0c;所有的作业都会通过 driver 编译 sql&＃xff0c;调度 job/task 执行&＃xff0c;分发 broadcast 变量&＃xff0c;二是对于每个 SQL&＃xff0c;相比于 HiveServer 会新起一个进程去处理这个 SQL 的执行&＃xff0c;STS 只有一个进程去处理&＃xff0c;如果某个 SQL 有异常&＃xff0c;查询了过多的数据量&＃xff0c; STS 有 OOM 退出的风险&＃xff0c;那么生产环境维持 STS 的稳定性就显得无比重要。

除了必要的存活报警&＃xff0c;首先我们区分了 ad-hoc 查询和离线调度的 STS 服务&＃xff0c;因为离线调度的任务往往计算结束时是把结果写入 table 的&＃xff0c;而 ad-hoc 大部分是直接把结果汇总在 driver&＃xff0c;对 driver 的压力比较大&＃xff1b;此外&＃xff0c;我们增加了基于 ZK 的高可用。对于一种类型的 STS&＃xff08;事实上&＃xff0c;有赞的 STS 分为多组&＃xff0c;如 ad-hoc&＃xff0c;大内存配置组&＃xff09;在 ZK 上注册一个节点&＃xff0c;JDBC 的连接直接访问 ZK 获取随机可用的 STS 地址。这样&＃xff0c;偶然的 OOM &＃xff0c;或者 bug 被触发导致 STS 不可用&＃xff0c;也不会严重到影响调度任务完全不可用&＃xff0c;给开发运维人员比较充足的时间定位问题。

权限控制

之后有另一个文章详细介绍我们对于安全和权限的建设之路&＃xff0c;这里简单介绍一下&＃xff0c;Hive 的权限控制主要包括以下几种:

• SQL Standards Based Hive Authorization
• Storage Based Authorization in the Metastore
• ServerAuthorization using Apache Ranger & Sentry

调研对比各种实现方案之后&＃xff0c;由于我们是从无到有的增加了权限控制&＃xff0c;没有历史负担。我们直接选择了 ranger &＃43; 组件 plugin 的权限管理方案。

除了以上提到的几个点&＃xff0c;我们还从社区 backport 了数十个 patch 以解决影响可用性的问题&＃xff0c;如不识别 hiveconf/hivevar (SPARK-13983)&＃xff0c;最后一行被截断 (HIVE-10541) 等等。

2.2 性能优化

之前谈到&＃xff0c;STS 只有一个进程去处理所有提交 SQL 的编译&＃xff0c;所有的 SQL Job 共享一个 Hive 实例&＃xff0c;更糟糕的是这个 Hive 实例还有处理 loadTable/loadPartition 这样的 IO 操作&＃xff0c;会阻塞其他任务的编译&＃xff0c;存在单点问题。我们之前测试一个上万 partition 的 Hive 表在执行 loadTable 操作时&＃xff0c;会阻塞其他任务提交&＃xff0c;时间长达小时级别。对于 loadTable 这样的 IO 操作&＃xff0c;要么不加锁&＃xff0c;要么减少加锁的时间。我们选择的是后者&＃xff0c;首先采用的是社区 SPARK-20187 的做法&＃xff0c;将 loadTable 实现由 copyFile 的方式改为 moveFile&＃xff0c;见下图&＃xff1a;

之后变更了配置 spark.sql.hive.metastore.jars&＃61;maven&＃xff0c;运行时通过 Maven 的方式加载 jar 包&＃xff0c;解决包依赖关系&＃xff0c;使得加载的 Hive 类是 2.1.1 的版本&＃xff0c;和我们 Hive 版本一致&＃xff0c;这样得好处是很多行为都会和 Hive 的相一致&＃xff0c;方便排查问题&＃xff1b;比如删除文件到 Trash&＃xff0c;之前 SparkSQL 删除表或者分区后是不会落到 Trash 的。

2.3 小文件问题

我们在使用 SparkSQL 过程中&＃xff0c;发现小文件的问题比较严重&＃xff0c;SparkSQL 在写数据时会产生很多小文件&＃xff0c;会对 namenode 产生很大的压力&＃xff0c;进而带来整个系统稳定性的隐患&＃xff0c;最近三个月文件个数几乎翻了个倍。对于小文件问题&＃xff0c;我们采用了社区 SPARK-24940 的方式处理&＃xff0c;借助 SQL hint 的方式合并小文件。同时&＃xff0c;我们有一个专门做 merge 的任务&＃xff0c;定时异步的对天级别的分区扫描并做小文件合并。

还有一点是 spark.hadoop.mapreduce.fileoutputcommitter.algorithm.version&＃61;2, MapReduce-4815 详细介绍了 fileoutputcommitter 的原理&＃xff0c;实践中设置了 version&＃61;2 的比默认 version&＃61;1 的减少了 70% 以上的 commit 时间。

三. SparkSQL 迁移之路

解决了大部分的可用性问题以后&＃xff0c;我们逐步开始了 SparkSQL 的推广&＃xff0c;引导用户选择 SparkSQL 引擎&＃xff0c;绝大部分的任务的性能能得到较大的提升。于是我们进一步开始将原来 Hive 执行的任务向 SparkSQL 转移。

在 SparkSQL 迁移之初&＃xff0c;我们选择的路线是遵循二八法则&＃xff0c;从优化耗费资源最多的头部任务开始&＃xff0c;把 Top100 的任务从 Hive 往 SparkSQL 迁移&＃xff0c;逐步积累典型错误&＃xff0c;包括 SparkSQL 和 Hive 的不一致行为&＃xff0c;比较典型的问题由 ORC 格式文件为空&＃xff0c;Spark 会抛空指针异常而失败&＃xff0c;ORC 格式和 metastore 类型不一致&＃xff0c;SparkSQL 也会报错失败。经过一波人工推广之后&＃xff0c;头部任务节省的资源相当客观&＃xff0c;在 2017 年底&＃xff0c;切换到 SparkSQL 的任务数占比 5%&＃xff0c;占的资源 20%&＃xff0c;资源使用仅占 Hive 运行的 10%-30%。

在 case by case 处理了一段时间以后&＃xff0c;我们发现这种方式不太能够扩展了。首先和作业的 owner 协商修改需要沟通成本&＃xff0c;而且小作业的改动收益不是那么大&＃xff0c;作业的 owner 做这样的改动对他来说收益比较小&＃xff0c;反而有一定概率的风险。所以到这个阶段 SparkSQL 的迁移之路进展比较缓慢。

于是我们开始构思自动化迁移方式&＃xff0c;构思了一种执行引擎之上的智能执行引擎选择服务 SQL Engine Proposer(proposer)&＃xff0c;可以根据查询的特征以及当前集群中的队列状态为 SQL 查询选择合适的执行引擎。数据平台向某个执行引擎提交查询之前&＃xff0c;会先访问智能执行引擎选择服务。在选定合适的执行引擎之后&＃xff0c;数据平台将任务提交到对应的引擎&＃xff0c;包括 Hive&＃xff0c;SparkSQL&＃xff0c;以及较大内存配置的 SparkSQL。

并且在 SQL Engine Proposer&＃xff0c;我们添加了一系列策略&＃xff1a;

• 规则策略&＃xff0c;这些规则可以是某一种 SQL pattern&＃xff0c;proposer 使用 Antlr4 来处理执行引擎的语法&＃xff0c;对于某些迁移有问题的问题&＃xff0c;将这种 pattern 识别出来&＃xff0c;添加到规则集合中&＃xff0c;典型的规则有没有发生 shuffle 的任务&＃xff0c;或者只发生 broadcast join 的任务&＃xff0c;这些任务有可能会产生很多小文件&＃xff0c;并且逻辑一般比较简单&＃xff0c;使用 Hive 运行资源消耗不会太多。
• 白名单策略&＃xff0c;有些任务希望就是用 Hive 执行&＃xff0c;就通过白名单过滤。当 Hive 和 SparkSQL 行为不一致的时候&＃xff0c;也可以先加入这个集合中&＃xff0c;保持执行和问题定位能够同时进行。
• 优先级策略&＃xff0c;在灰度迁移的时候&＃xff0c;是从低优先级任务开始的&＃xff0c;在 proposer 中我们配置了灰度的策略&＃xff0c;从低优先级任务切一定的流量开始迁移&＃xff0c;逐步放开&＃xff0c;在优先级内达到全量&＃xff0c;目前放开了除 P1P2 以外的 3 级任务。
• 过往执行记录&＃xff0c;proposer 选择时会根据历史执行成功情况以及执行时间&＃xff0c;如果 SparkSQL 效率比 Hive 有显著提升&＃xff0c;并且在过去一直执行成功&＃xff0c;那么 proposer 会更倾向于选择 SparkSQL。

截止目前&＃xff0c;执行引擎选择的作业数中 SparkSQL 占比达到了 73%&＃xff0c;使用资源仅占 32%&＃xff0c;迁移到 SparkSQL 运行的作业带来了 67% 资源的节省。

未来展望

我们计划 Hadoop 集群资源进一步向 SparkSQL 方向转移&＃xff0c;达到 80%&＃xff0c;作业数达 70%&＃xff0c;把最高优先级也开放到选择引擎&＃xff0c;引入 Intel 开源的 Adaptive Execution 功能&＃xff0c;优化执行过程中的 shuffle 数目&＃xff0c;执行过程中基于代价的 broadcast join 优化&＃xff0c;替换 sort merge join&＃xff0c;同时更彻底解决小文件问题。

作者简介&＃xff1a;大数据平台是有赞共享技术的核心团队之一&＃xff0c;该团队主要由数据技术、数据产品、算法挖掘、广告平台四个小团队组成&＃xff0c;目前共有 34 位优秀的工程师组成。

今日荐文

点击下方标题即可阅读

Kafka 背后公司获 1.25 亿融资&＃xff0c;估值超 25 亿美元

面向21W&＃43;AI爱好者、开发者和科学家&＃xff0c;每周一节免费AI公开课&＃xff0c;囊括上万人的AI学习社群&＃xff0c;提供最新AI领域技术资讯、一线业界实践案例、搜罗整理业界技术分享干货、最新AI论文解读。回复「AI前线」、「TF」等关键词可获取干货资料文档。

如果你喜欢这篇文章&＃xff0c;或希望看到更多类似优质报道&＃xff0c;记得给我留言和点赞哦&＃xff01;╰(&＃xffe3;ω&＃xffe3;&＃xff4f;)