Apache DolphinScheduler 助力 Trino 快速实现湖仓一体数据建设

海豚调度 2023-02-23

756

点亮 ⭐️ Star · 照亮开源之路

https://github.com/apache/dolphinscheduler

作者 | 钟嘉杰 Apache DolphinScheduler

PMC

Member

在面对联合查询和湖仓一体场景时，Trino 已经是不少开发者的不二之选。Trino 是一个查询引擎，在数仓、即席查询方面非常强大。然而，在整个数仓处理场景中，相信很多数仓用户关注的不仅是查询，还有应用层面的可视化编辑 SQL 能力、失败重试、任务并发控制、SQL 版本管理、任务编排等能力，但这些 Trino 还未提供。幸运地是，Apache DolphinScheduler 作为一个工作流调度平台，最擅长的就是提供这样的能力。如果两者相结合，就可以互相取长补短，补足彼此的能力。

本文中，我们将通过实例，分析 Apache DolphinScheduler 如何与 Trino 结合，满足用户实现 OLAP 业务的需求，同时引入尽可能少的组件，快速实现湖仓一体数据建设。

什么是 Trino

Trino是一种开源分布式SQL查询引擎，旨在查询分布在一个或多个异构数据源上的大型数据集。(Trino is a query engine that runs at ludicrous speed, Fast distributedSQLquery engine for big data analytics that helps you explore your data universe.) 虽然 Trino 可以理解 SQL，但是其使用场景并不是我们常见的Online Transaction Processing (OLTP)，大家更多地将它用于在湖仓一体中处理 Terabytes or Petabytes 级别的数据。因为它同时支持湖仓一体, 数据分析和 Ad-hoc 场景，很多公司将其作为从处理源数据，构建湖仓一体开始，到终端用户的 ad-hoc 结束的全链路工具。

Trino 具有强大的数据处理能力，允许多台机器同时协作，与 DolphinScheduler 一样，Trino 支持高可用和 connectors 插件化，能保障终端服务的可用性，有较高的水平扩展性。

什么是 DolphinScheduler

Apache DolphinScheduler 作为 Apache 基金会的一个 Top Level Project，是一个分布式和可扩展的开源工作流协调平台，具有强大的DAG可视化界面。其最主要的特点是：平台化, 拖拉拽, HA, 高性能, 以及插件化设计。

平台化：自动处理非业务的逻辑，如失败重试，并发控制，检控，复杂任务依赖等，让用户更加容易使用
拖拉拽：通过 UI 创建 workflow 首要地位，同时支持 API 创建，让 engineer 和 not-engineer 都能很好的使用
HA：Decentralization，原生 HA 队列，保障集群可用性
插件化设计：简单方便地扩展插件，更方便地和用户系统集成，同时有很多内部插件

用户场景

我们将通过一个实际的例子，简述如何通过 Trino 和 DolphinScheduler 的结合，满足用户实现 OLAP 业务的需求，同时引入尽可能少的组件。

公司发展初期 – OLTP阶段

Trino 作为一个适用于湖仓一体和 ad-hoc 的查询引擎，非常适用于快速发展的公司，可以只维护同一套 infra 让数据业务运行起来。其常见的使用场景是用户的业务系统有数个 RDBMS ，里面存储了用户常见的业务数据，包括用户基本信息、订单信息、商品信息等，另一个 RDBMS 储存了商家信息，商品销售情况等。另外，企业还需要从外部获取部分数据，如行政区划分，各个区的GDP，各个地区最新政策，以及区域人口信息等。

公司成立初期，用户专注于业务系统开发，这个时候大部分的操作还是基于 RDBMS 的 OLTP 操作，即对 RDBMS 进行 CURD，完成特定的业务动作。

公司发展中期 – OLAP阶段

但是随着业务的发展，用户希望通过 BI 工具展现数据，用数据指导业务活动，促进用户成单，增加销售额，以及通过对销售期间商品数据的分析，得出各个活动的 ROI (Return On Investment) 。这时候用户需要一个基于分析的系统，原来的 RDBMS 数据库的方式已经不足以支撑类似的需求了。

用户此时想要有一个基础组件，能同时满足定时的 BI 报表，又能让分析师，产品可以做快速查询的工具。这样，他们只需要维护同一套 infra 设施就能让业务跑起来。Trino 就是这样一个工具，其高性能、高稳定性、多 connector 的特性，能同时满足用户批量离线数据的需求，保证 ad-hoc 查询的实效性。

此时，用户的 OLAP 系统数据流向是这样的，Trino 会读源数据库 RDBMS 和外部系统的数据，在 Trino 中做构建湖仓一体，同时让数据支持两个系统，BI visualization（对于部分常规固定报表的统计）和 Analysis APP（探索性、实验性查询）。

SELECT count(*), mktsegment, nationkey,
       CAST(sum(acctbal) AS bigint) AS totalbal
FROM customer
GROUP BY mktsegment, nationkey
HAVING sum(acctbal) > 5700000
ORDER BY totalbal DESC;

面临的问题

用户在构建湖仓一体时，会遇到各种问题。下面我们将围绕这些问题的阐述，以及 Apache DolphinScheduler 是如何解决这些问题而展开。

许多零散SQL管理问题

在这个分析系统创建期间，面临最大的问题是湖仓一体的构建。为了更好地节省数据计算的时间，降低SQL的复杂度，人们一般会将湖仓一体构建成拥有多个 Layer 的系统，并产生多个中间表。将上图的 Trino 进行拆解，就会发现它由很多SQL组成，并且SQL之间有不同的联系。如果将每个SQL任务作为一个节点，SQL任务间先后执行的关系作为一条线，那SQL任务本身及其关系就构成了一个有向无环图（DAG）。在湖仓一体初期，用户可通过 Crontab 任务按照一定的时间执行不同的 SQL，或者通过 Python 等将SQL任务链接起来。

但是数据的价值巨大，当有了湖仓一体并从分析中得到了正向的结果后，就需要探索更多的数据，这意味着更多的分层，分层增多会给原来的 Crontab 或者 Python 任务带来压力，有时候找到合适的位置将SQL任务放入 DAG 甚至比编写SQL任务的所花的时间还要多。

可能会有多条 SQL

INSERT INTO customer_total_bal
SELECT count(*), mktsegment, nationkey,
       CAST(sum(acctbal) AS bigint) AS totalbal
FROM customer
GROUP BY mktsegment, nationkey
HAVING sum(acctbal) > 5700000
ORDER BY totalbal DESC;

INSERT INTO package_addr_weight
SELECT origin_state, destination_state, origin_zip, sum(package_weight)
FROM shipping
GROUP BY DISTINCT
    CUBE (origin_state, destination_state),
    ROLLUP (origin_state, origin_zip);

INSERT INTO clerk_price_rnk
SELECT orderkey, clerk, totalprice,
      rank() OVER w AS rnk
FROM orders
WINDOW w AS (PARTITION BY clerk ORDER BY totalprice DESC)
ORDER BY count() OVER w, clerk, rnk

Apache DolphinScheduler 提供了简单高效的 web UI，用户可以用 DolphinScheduler 管理SQL脚本，管理方式包括直接在SQL任务定义管理，或者资源中心管理。

SQL任务定义管理

在 DolphinScheduler 的 datasource 中创建 Trino 的数据源

回到 Project 页面创建一个SQL任务，并将上述的SQL脚本粘贴到SQL任务定义中，即可通过 DolphinScheduler 管理SQL脚本。DolphinScheduler 还能为每个任务提供SQL脚本的版本管理，方便用户做版本回退。

资源中心管理SQL文件

如果你倾向于使用外部储存，如 Amazon S3，MinIO 或者 HDFS 来保存你SQL脚本，然后通过 Command line interface 的方式触发，你可以利用 DolphinScheduler 的资源中心，以及 Shell 任务类型完成这一切。使用资源中心之前，你需要先配置并启动 DolphinScheduler，可以参考这里完成配置。启动资源中心后，可以将本地的脚本通过 DolphinScheduler 资源中心上传到对应远端存储介质。

在通过 DolphinScheduler 资源中心管理后，我们可以在 DolphinScheduler 中创建一个 Shell 任务，在任务中引用对应的SQL脚本，并通过 Trino Command line interface 的方式执行对应的SQL。需要注意的是，当你选择使用这个方式提交任务的时候，DolphinScheduler 每个 worker 节点都需要安装 Trino Command line interface 并将其加到 $PATH 中。

SQL依赖问题

除了SQL脚本的管理问题外，各个SQL任务之间的依赖也是将要解决的问题。湖仓一体的多层结构，意味着有部分SQL需要先运行，有部分SQL需要后运行，那么任务执行的顺序，以及各个任务之间的依赖就显得非常重要。

DolphinScheduler 作为一个任务调度系统，天然地能够解决复杂任务依赖的问题。DolphinScheduler 编排工作流的方式遵循 Drag and Drop First 的原则，你可以通过在 web UI 页面进行 Drag and Drop 的方式建立、删除任务之间的关系，你需要做的就是通过鼠标，将前后两个任务链接起来。

DolphinScheduler 的版本管理，不仅能管理工作流中各个任务的定义，还能管理任务依赖，意味着我们在做版本 revert 的时候，能将拿到对应版本整个工作流的信息。

湖仓一体构建过程中，处理上述同一工作流中直接的任务依赖外，还会有特殊的依赖，DolphinScheduler 均提供了 out-of-box 的解决方式，如：

当当前工作流的特定任务成功时，想要触发下游工作流运行，可以使用 DolphinScheduler 中的任务程序
不仅想要基于当前工作流上游任务触发任务，还希望触发其他工作流中的任务，选择任务依赖设置不同工作流中的上游任务
还想在上游任务失败而不是成功时触发一些下游任务，可以根据上游任务状态根据上游任务状态尝试不同的下游任务条件和路线。

Connector 系统压力大导致查询慢或者失败

使用 Trino 过程中，用户和 connector 打交道应该是最多的。Trino 有一个优势，允许一个SQL脚本对来自不同 connector 的数据进行操作。这大大降低了用户使用成本，用户不再需要将数据从一个地迁移到另一个地方，然后再做关联查询。但是直接对原数据库做查询也可能会面临一些问题，比如源数据库是业务数据库，当业务高峰期的时候，数据库读的速度会变慢，如果SQL正好对这个数据库进行操作，就会面临查询变慢，甚至失败的问题。

如何保证 Connector 查询慢甚至失败的时候，SQL任务仍然能正常产出，或者及时发送 Alert 消息给用户显得尤为重要。

针对查询慢问题，DolphinScheduler 提供开箱即用的 Timeout 机制，允许用户为单个任务配置 Timeout Threshold，当任务运行时间超过了Timeout period 还没有运行成功时，将会触发 Timeout Strategy，Timeout Strategy 可以选择近发送 Alert 告警，或者发送 Alert 告警的同时，标记任务为失败状态。这样，用户可以更好地运维湖仓一体中的任务。

可能你会问了，标记任务失败之后，是否需要人为登陆DolphinScheduler系统重新运行任务呢？有没有更加智能的方式让任务失败之后可以自动被拉起呢？答案是 DolphinScheduler 的失败重试策略，你可以指定任务失败后重试的次数和间隔，如这个例子中，我们配置了5次重试次数，每次重试间隔为 10 分钟，这样做能有效减少源数据库有压力时候人为介入处理的次数，让工作流运行更加流畅。当任务失败的次数被超过了设置的5次后，任务将会被标记失败，并且不会重新运行。

大量任务并发调度

Trino 可以支持大量SQL并发运行，我们自然不希望引入了workflow orchestration 并编排了调度之后会影响到 Trino 的并发数，这就要求 workflow orchestration 工具可以支持多任务并发，DolphinScheduler 是一个无中心的架构，任务调度核心流程相关的 Master 和 Worker 都支持水平扩展，保证单位时间内运行更多的任务，经过测试 DolphinScheduler 3.0.2 可以在单台 8c16g 机器上每秒运行 40 个任务。

便于部署定时

用户完成湖仓一体任务的编排后，一般需要为不同的 workflow 配置不同的调度周期，保证 workflow 可以按照需要的频率新增或更新数据，DolphinScheduler 可以在 UI 上完成 workflow 定时配置，只需要在 workflow definition 中找到对应的工作流，点击 timing，配置调度的 start time， end time，并配置 crontab 表达式，和时区，就能完成配置。当 workflow 有了定时后，DolphinScheduler 将在定时开始时启动 workflow 的运行。

补数

当完成 workflow 的编排并测试了基础功能后，我们再回到湖仓一体场景中，因为是面向历史的数据分析，所以不仅需要分析增量数据，还需要有历史全量数据。这时候，根据增量场景的代码完成数据 backfill 显得非常重要，我们不希望为增量数据和全量数据写两套代码，而是希望用同一套增量代码，通过不同的运行时间配置完成历史数据 backfill。湖仓一体提供了 Complement Data 的功能完成 data backfill。

如图，只需要在 trigger 选择 Complement Data 功能，选择backfill 的时间段，已经 backfill 的策略，点击 comfirm 就能完成 backfill 任务的触发。等全部的 backfill 任务运行完成，该时间段的历史数据也就全部就位了，如图中当任务完成的时候，我们就完成 2019-01-01 到 2022-01-01 历史数据的 backfill。

任务实例

DolphinScheduler 会保留每个 workflow 以及 task 的运行状态和记录，让用户更好地查找问题，并重跑 workflow 或者 task。意味着当你对 Trino 计算的数据有疑问时，你可以通过 workflow instance 或者 task instance 向前回溯 workflow。