数据库
首页 > 数据库> > 看SparkSql如何支撑企业数仓

看SparkSql如何支撑企业数仓

作者：互联网

企业级数仓架构设计与选型的时候需要从开发的便利性、生态、解耦程度、性能、 安全这几个纬度思考。本文作者：惊帆 来自于数据平台 EMR 团队

前言

Apache Hive 经过多年的发展，目前基本已经成了业界构建超大规模数据仓库的事实标准和数据处理工具，Hive 已经不单单是一个技术组件，而是一种设计理念。Hive 有 JDBC 客户端，支持标准 JDBC 接口访问的 HiveServer2 服务器，管理元数据服务的 Hive Metastore，以及任务以 MapReduce 分布式任务运行在 YARN 上。

标准的 JDBC 接口，标准的 SQL 服务器，分布式任务执行，以及元数据中心，这一系列组合让 Hive 完整的具备了构建一个企业级数据仓库的所有特性，并且 Hive 的 SQL 服务器是目前使用最广泛的标准服务器。

虽然 Hive 有非常明显的优点，可以找出完全替代 Hive 的组件寥寥无几，但是并不等于 Hive 在目前阶段是一个完全满足企业业务要求的组件，很多时候选择 Hive 出发点并不是因为 Hive 很好的支持了企业需求，单单是因为暂时找不到一个能支撑企业诉求的替代服务。

企业级数仓构建需求

数仓架构通常是一个企业数据分析的起点，在数仓之下会再有一层数据湖，用来做异构数据的存储以及数据的冷备份。但是也有很多企业，特别是几乎完全以结构化数据为主的企业在实施上会把数据湖和企业数仓库合并，基于某个数仓平台合二为一。

企业在考虑构建自身数仓体系的时候，虽然需要参考现有的行业技术体系，以及可以选择的组件服务，但是不能太过于局限于组件本身，寻找 100%开箱即用的产品。太过于局限于寻找完全契合的组件服务必然受限于服务本身的实现，给未来扩展留下巨大的约束。企业数据仓库架构必然不等于一个组件，大部分企业在数仓架构实施的都是都是基于现有的部分方案，进行基于自己业务合适的方向进行部分开发与定制，从而达到一个半自研的稳态，既能跟上业务变化的速度，又不过于依赖和受限于组件自身的发展。

一般来说企业级数仓架构设计与选型的时候需要从以下几个纬度思考：

对于企业数仓架构来说，最重要的是如何基于企业业务流程来设计架构，而不是基于某个组件来扩展架构。

一个企业数仓的整体逻辑如上图所示，数仓在构建的时候通常需要 ETL 处理和分层设计，基于业务系统采集的结构化和非结构化数据进行各种 ETL 处理成为 DWD 层，再基于 DWD 层设计上层的数据模型层，形成 DM，中间会有 DWB/DWS 作为部分中间过程数据。

从技术选型来说，从数据源的 ETL 到数据模型的构建通常需要长时任务，也就是整个任务的运行时间通常是小时及以上级别。而 DM 层主要是支持业务的需求，对实效性要求比较高，通常运行在 DM 层上的任务时间在分钟作为单位。

基于如上的分层设计的架构图可以发现，虽然目前有非常多的组件，像 Presto，Doris，ClickHouse，Hive 等等，但是这些组件各自工作在不同的场景下，像数仓构建和交互式分析就是两个典型的场景。

交互式分析强调的是时效性，一个查询可以快速出结果，像 Presto，Doris，ClickHouse 虽然也可以处理海量数据，甚至达到 PB 及以上，但是主要还是是用在交互式分析上，也就是基于数据仓库的 DM 层，给用户提供基于业务的交互式分析查询，方便用户快速进行探索。由于这类引擎更聚焦在交互式分析上，因此对于长时任务的支持度并不友好，为了达到快速获取计算结果，这类引擎重度依赖内存资源，需要给这类服务配置很高的硬件资源，这类组件通常有着如下约束：

一旦发生任务异常，例如网络抖动引起的任务失败，机器宕机引起的节点丢失，再次重试所消耗的时间几乎等于全新重新提交一个任务，在分布式任务的背景下，任务运行的时间越长，出现错误的概率越高，对于此类组件的使用业界最佳实践的建议也是不超过 30 分钟左右的查询使用这类引擎是比较合适的。

而在离线数仓场景下，几乎所有任务都是长时任务，也就是任务运行时常在小时及以上，这时就要求执行 ETL 和构建数仓模型的组件服务需要具有较高的容错性和稳定性，当任务发生错误的时候可以以低成本的方式快速恢复，尽可能避免因为部分节点状态异常导致整个任务完全失败。

可以发现在这样的诉求下类似于 Presto，Doris，ClickHouse 就很难满足这样的要求，而像 Hive，Spark 这类计算引擎依托于 Yarn 做资源管理，对于分布式任务的重试，调度，切换有着非常可靠的保证。Hive，Spark 等组件自身基于可重算的数据落盘机制，确保某个节点出现故障或者部分任务失败后可以快速进行恢复。数据保存于 HDFS 等分布式存储系统上，自身不管理数据，具有极高的稳定性和容错处理机制。

反过来，因为 Hive，Spark 更善于处理这类批处理的长时任务，因此这类组件不擅长与上层的交互式分析，对于这种对于时效性要求更高的场景，都不能很好的满足。所以在考虑构建数仓的时候，通常会选择 Hive，Spark 等组件来负责，而在上层提供交互式分析查询的时候，通常会使用 Presto，Doris，ClickHouse 等组件。

归纳下来如下：

一个企业在实施数据平台的时候，由多个不同组件各自工作在不同的架构层中，无法相互取代，相互协作配合，承载整个企业的数据平台业务。

企业级数仓技术选择

Google 发表的三篇论文从存储，计算，检索三个方向阐述了海量数据下一种新的分布式数据加工处理技术，这三个方向被雅虎 Nutch 团队实现后贡献给 Apache，也就是目前大家看到的 HDFS，MapReduce 和 HBase，形成了早期 Hadoop 的三大利器。

然而这三大利器更聚焦在异构数据的信息提取处理上，没有提供对结构化数据很友好的类似 SQL 语法的分析入口，同时在编程态的支撑也不够友好，只有 Map 和 Reduce 两阶段，严重限制了业务处理的实现，雅虎团队也是爬虫相关业务孵化而出，可以看出 Hadoop 早期的三大套件有着如下特点：

Hive 就是诞生在这样的较大的行业背景下，Hive 的出现刚好弥补了 Hadoop 只能用来做离线数据处理这个缺陷，提供了一种常用的分析接口，并且提供了非常好的用户交互方式。

Hive 整体架构如上图所示（本图来自于 Hive 官网），Hive 提供 JDBC 接口实现支持以编程形式进行交互，同时业内几乎所有 SQL Client、开源或商业 BI 工具都支持通过标准 JDBC 的方式连接 Hive，可以支持数据探索的动作，极大的丰富了大数据生态圈下的组件多样性，同时也降低了使用门槛，可以让熟悉 SQL 的人员低成本迁移。

基于这些设计非常好的特效，加上 Hive 经过这多年的逐步完善，发展到今天已经是一个非常稳定成熟的生产环境可用的数据仓库组件，甚至替代品都很难找到，因此使用 Hive 作为数据仓库的构建基础是一个非常好的选择。

如上图所示，其中有很多优点：

所以虽然 Hive 出现已经非常有很长时间了，但是依旧是数仓构建的首选，在整个数仓构建中随处可见 Hive 的身影。虽然 Hive 有种种优点，让人难以割舍，但是并不等于能很好的支撑企业业务需求。很多时候选择 Hive 仅仅是因为暂时没有其他可选的组件，如果自己从头开发一个，或者基于某个组件改造，成本又会远超企业预期，因此不得不继续选择使用 Hive。

基于实践来看，Hive 在构建企业数仓过程中存在的主要局限围绕在以下几个方面：

虽然 Hive 在以上局限层面也做了很多尝试，Hive On Spark，但是受限于 Hive 的架构，HiveServer2 自身有自己的 SQL 解析引擎，为了兼容架构将解析后的结果直接翻译成 Spark 最底层的接口，整体性能反而提升不大。

除了 Hive 之外，还有非常多的其他优秀的组件，但是从企业数仓技术选型的视角来看，适合用来构建数据仓库的，目前只有 Hive 和 Spark SQL 相对更加合适，在这两个组件中，Spark SQL 相对 Hive 的优势又更加明显。

SparkSQL 如何支撑企业级数仓

Spark 引擎因为自身强大的生态和方便的编程接口被广泛应用在数据处理场景下，Spark 提供的 Spark SQL 模块更是将使用 Spark 支撑企业数据仓库提供了一个良好的基础设施。

如上图所示，一个典型的数据仓库架构需要包含不同层次的模型构建。由于数据量大，数据结构异构等多种原因，大数据架构下的企业数仓构建抛弃了基于关系型数据库下的 Cube 设计，直接采用基于分布式任务进行处理来构建多层数据模型。因此对于构建企业数仓的服务来说，有着如下要求：

基于以上特性可以发现，在目前可选择的组件范围内，Spark SQL 相比其他组件，乃至 Hive 更加合适承担这类任务。但是很多企业在进行架构设计的时候割舍不掉 Spark SQL 带来的丰富特性，又愁于 Spark SQL 缺乏类似 Hive 这样的 SQL 服务器，于是退而求其次变成 Hive 与 Spark SQL 两个组件共存的形态，Hive 退化为仅仅提供 MetaStore 服务，因此从很多实践的现象来看，Hive 构建企业数仓已是过去式，采用 Spark SQL 进行数据仓库的构建是众多的选择。

如上图所示，企业在构建数仓的时候，通过一个 Spark SQL Server 提供基于 SQL 接口的常驻服务，同时也可以采用 Spark Submit 的方式直接提交 Jar 任务去运行，既能达到提供标准 SQL 交互式接口，又能提供更灵活的编程态接口。

从不同的企业级数仓构建视角来看，Hive 带来的约束都越来越大，而 Spark SQL 的成熟度和发展趋势已经完全具备取代 Hive 来构建整个数仓，Spark SQL 的优势集中体现在如下方面：

因此，完全基于使用 Spark SQL 来支撑企业级的数仓是完全可行的，并且在目前也被众多企业实践验证。

如上图所示，一个基于 Spark SQL 构建的企业数仓架构逻辑架构设计上包含以上几个部分，每一个 Spark SQL 引擎都是一个服务器，Spark SQL 引擎将自己的信息注册到 Zookeeper 中，SQL 服务器基于 Zookeeper 中的 Spark SQL 引擎来执行客户端过来的请求，SQL 服务器是一个兼容 Hive JDBC 接口的服务器，在使用 Spark SQL 来支撑数仓构建的时需要重点考虑的实施点是：

使用 Spark SQL 支撑企业级数仓的核心的地方还是在于如何提供一个好用的任务服务器，用来支撑任务的管理。任务管理服务器在逻辑上与 HiveServer2 相似，但是更加的轻量，没有 HiveServe2 中复杂而繁重的 SQL 解析，同时又没有 Spark Thrift Server 这种自身就是一个 YARN 作业的约束。企业可以基于自身的业务流程，开发一个轻量的服务器，在这方面字节有非常深的实践经验，同时也有自己的 Spark SQL 引擎服务器，可关注后续的动态。同时业界也有其他企业做了类似的工作，例如网易开源的 Kyuubi。

Kyuubi 整个架构图如上所示（图片来自于 kyuubi 官网：https://github.com/apache/incubator-kyuubi）。

Kyuubi 基于 Spark SQL 之上，较好的弥补了 Spark Thrift Server 在多租户、资源隔离和高可用等方面的不足，是一个真正可以满足大多数生产环境场景的开源项目。但是 Kyuubi 在设计的时候考虑的是如何弥补 Spark Thrift Server 的不足，目的在于增强 Spark SQL 的能力，而不是对等设计一个可以替换 Hive 组件的服务。因此对于遗留项目来说迁移成本较高，Spark SQL 与 Hive 有着两套不兼容的 SQL，在使用 Kyuubi 的时候如何是遗留系统迁移成本将是一个非常大的工作量。

而行业也有开源的 Spark Thrift Server，该思路是非常优秀的，但是因为开发过程中有点太过于局限，导致依旧存在很多问题，主要体现在：

因此虽然 Spark 提供了 Spark thrift server 服务用来提供类似 JDBC 这样的接口交互方式，但是目前依旧缺乏很多生成功能，导致在生产环境使用的情况非常少，Spark thrift server 更像是一个小众的半成品，小修小补的尝试着解决部分问题，但是没有给予一个彻底的方案，导致现在有点缺乏实际的生产应用。

字节跳动在 Spark SQL 的优化实践

字节跳动的数仓在 2020 年全面从 Hive 迁移至 Spark SQL，除了组件层面的源码修改外，在使用层面也对 Spark 做了非常多的优化。

数据湖引擎集成

Hudi，Iceberg 等数据湖引擎目前使用的越来越广泛，字节内部在使用 Spark SQL 的时候也存在需要使用数据湖引擎的需求，因此需要将数据湖引擎集成到 Spark SQL 中，这个过程碰到非常多的问题。

首先在与 Iceberg 集成的时候，对体验和易用的问题进行了优化，用户在使用 Spark SQL 过程中，需要手动输入很多指令，并且需要找到对应的 spark-iceberg 依赖包，这个也是目前集成 Iceberg 最常用的方案。我们的解决方式是在预先安装的过程中，提前把 iceberg 的相关 jar 包放到 spark jars 目录下，这样用户只需要指定 catalog 即可，无需再手动输出很多指令。

其次在 Spark 与 Hive 跨引擎分析场景下使用 Iceberg，Spark 正常创建表，Presto/Trono 可以正常读写，但 Hive 无法正常读写，这个问题官方的文档也没有清晰的描述，解决方案是需要修改 Spark 的配置文件或者修改 Hive 的 hive-site-spark override 配置，确保初始化出来的 Spark Session 中的配置项 iceberg.engine.hive.enable 的值为 true，Hive 才能正常的读取 Spark 创建的表。

问题上本质上是由于 Iceberg 为了支持 Hive 引擎，在整体的设计上做了一些妥协，使用了 Storage Handler 的方式去实现 Hive 对 Iceberg 格式的表的读写，需要显式的指定 Hive 的 Input/Output Format 实现，而 Presto/Trono 则可以基于 Hive 的 format_type 自动识别表的格式进行识别。

在兼容性上，由于 Iceberg 0.12 版本不支持 Spark 3.2，由于升级 Spark 的影响范围非常大，于是更新了 Iceberg，使用了社区的一个 master 的 snapshot 版本进行编译，与 Spark 3.2 进行集成。

Spark SQL 服务器

虽然行业针对 Spark SQL 提供一个 SQL 服务器已经有 Spark Thrift Server 或者 Kyuubi 这样的工具，但是在字节自身业务的背景下，这些工具并不能完全满足要求，因此自己也设计实现了自己 Spark SQL Server，主要聚焦解决的是如下场景：


如上图所示，SQL 服务器是一个实现了 Thrift 接口的服务器，提供标准的 JDBC 访问接口，Spark SQL 引擎同样实现了 Thrift 接口，Spark SQL 引擎在服务启动的时候便已经被提交至 Yarn，处于等待状态。当业务任务到达的时候，由 SQL 服务器实现引擎的筛选，匹配一个已经存在的引擎，或者重新提交一个全新的引擎用来执行任务。

SQL 服务器支持 OpenLDAP，Kerberos 等常用的权限认证，同时支持多种不同的隔离级别，例如 Session 级别则每一个业务 SQL 都会初始化一个 Spark SQL 引擎用来接收任务，任务执行结束后，引擎从 Yarn 中销毁。而 User 级别则针对用户会初始性 0-N 个引擎，常驻于 Yarn 中，处于交替执行任务。

这样的服务器设计打破了 Spark Thrift Server 的单 Driver 所带来的局限，解耦了 SQL 服务和任务执行。也就支持更细粒度的资源管理和跨队列的任务提交。

同时也兼容了 Hive 的接口，用户可以通过如下方式访问服务器：

HA 访问链接：

./bin/beeline -u jdbc:hive2://emr-5fqkwudj144d2gc1k8hi-master-1/;serviceDiscoveryMode=zooKeeper;zooKeeperNamespace=midas/ha;auth=LDAP -n emr_dev -pEMR123456emr

非 HA 访问链接：

./bin/beeline -u jdbc:hive2://emr-master-2:10005/default;auth=LDAP” -n test_sub -pEMR123456emr

HA 模式下的信息被记录在 Zookeeper 中，保存的内容格式与 HiveServer2 的内容一致，能确保使用 Hive 的客户端可以直接访问 HA 模式下的服务器。

Spark SQL 多租户


在 Hive 任务执行过程中，HiveServer2 服务承担了提供 SQL 服务器进行用户身份认证，权限判断，以及解析 SQL 生成最终的执行计划，再由 MR 引擎执行具体的分布式任务。

在这个过程中 HiveServer2 承担了非常重的职责，因此需要消耗非常大的资源，因此会很大程度的影响用户的并发。对于分布式任务运行来说，它的资源约束来自于 Yarn 作为资源管理器所分配的资源，但是在 Hive 架构下却受限于 HiveServer2 的影响，导致用户并发的数量无法随着 Yarn 资源的提升进行提升。

而在 Spark SQL 引擎中，SQL 解析是下推到引擎内部，与具体的分布式任务执行合为一体，不需要单独的服务器去做 SQL 解析。也正因为 Spark SQL 与 Hive 在解析模块的架构存在差异，Hive On Spark 的模式会变得非常难。

针对如上的场景，字节跳动重新设计的 SQL 服务器只负责任务的接收，进行用户资源，权限和身份的判断，然后将任务发送给运行在 Yarn 中的 Spark SQL 服务器。打破了 Hive 这种并发受制于 HiveServer2 和 Yarn 两层约束的局面，只由 Yarn 的资源决定用户的并发程度，从而极大的降低了 Spark SQL 服务器的资源需求，增强了其稳定性，在用户并发上有了非常大的提升。

其次通过引擎预热的功能减少任务执行的时间，提升整体速度，引擎预热指的是在服务启动的时候便向 Yarn 提交 Spark SQL 引擎，处于等待的状态，当业务请求到达的时候，基于业务类型从已经处于就绪的引擎中选择一个引擎来执行任务。

并不是每一个预热提交的引擎都会被选择执行，在 SQL 服务器中存在如下三种引擎隔离级别：

由此可见只有在 User，Open 的级别下引擎预热才会产生价值，预热省去了 Spark Submit 的时间，当用户数量足够多，群体为统计单位所节省的时间越多。

Spark SQL 事务支持

Hive 的事务力度是基于 HiveServer2 实现的，例如通过如下语法：

CREATE TABLE tm (a int, b int) stored as orc TBLPROPERTIES ('transactional'='true', 'transactional\_properties'='insert\_only')

可开启事务，但是由于对事务的管理是在服务器上，因此需要开启 ACID 的时候受影响的是整个 HiveServer2 的所有请求，而 Spark SQL 很好的集成和支持了 Hudi，Iceberg 等数据湖格式，因此在 Spark SQL 服务器中不需要实现类似 HiveServer2 的事务机制，只需要在最终读取处理数据的时候，采用 Hudi，Iceberg 等特性便可达到支持事务的效果。

例如对于 Icdberg 数据格式的表已支持 update、delete 操作：

MERGE INTO prod.nyc.taxis ptUSING (SELECT * FROM staging.nyc.taxis) stON [pt.id](http://pt.id) = st.idWHEN NOT MATCHED THEN INSERT *;

因此当 Spark SQL 集成了 Iceberg 后，便具有了事务能力，再配合 SQL 服务器，便可在很低的成本上具有和 Hive 事务能力同等的效益，同时也没有 Hive 下的一些约束，从这样的架构设计上来看，能够完整的替换 Hive。另外一个方面当用户数量变多，同时数据会发生修改，更新等操作，很容易造大量的小广播传输，从而引起 Driver 的 OOM。虽然大广播也会存在 OOM 的问题，但是大广播可以通过阈值控制，而小广播阈值对其不生效，一旦说数量变多，很容易引起 Driver 的 OOM。字节通过对小广播进行合并广播，解决大量小广播进行传播，导致打爆 Driver 的情况出现。

尾声

随着企业的业务发展越来越复杂，需要更加灵活，更加高效的数仓架构，在这样的业务驱动背景下，Hive 的局限变得越来越明显，而基于 Spark SQL 灵活构建数仓的方案将会变得越来越主流。所以企业在考虑数据仓库构建体系的时候，可以考虑如何基于 Spark SQL 构建自身数据体系，Spark 完善和开放的生态在未来必然会有更多优秀的服务会围绕 Spark 形成强大的优势。

我们基于企业级数仓建设的部分经验技术也已经通过火山引擎E-MapReduce对外开放。

欢迎关注字节跳动数据平台同名公众号

标签：数仓,Hive,任务,引擎,SparkSql,SQL,Spark,支撑
来源： https://www.cnblogs.com/bytedata/p/15918335.html