实现跨库连表 SQL 生成器

一 概述

ADC（Alibaba DChain Data Converger）项目的主要目的是做一套工具，用户在前端简单配置下指标后，就能在系统自动生成的大宽表里面查询到他所需要的实时数据，数据源支持跨库并支持多种目标介质。说的更高层次一点， 数据的全局实时可视化这个事情本身就是解决供应链数据“神龙效应”的有效措施（参考施云老师的《供应链架构师》[1]一书）。做ADC也是为了这个目标，整个ADC系统架构如下图所示：

架构解析：

• 初始数据来自于元数据中心。


• 经过元数据适配层后转换为内部格式数据。

• 调度中心把内部格式的数据传到计划中心，计划中心分析数据需求并建模，通过SQL生成器生成资源和SQL，分别通过告警中心、对账中心设定监控标准和对账标准。


• 对账中心定时对账，查看数据的对齐情况。
• 告警中心可以针对任务错误、延迟高等情况发送报警。

• 资源的生命周期管控在资源管理中心下，view删除时资源管理中心负责回收资源。


• 基础资源适配层主要借助集团基础资源管理能力串联阿里各类数据服务， 比如阿里云MaxComputer、Flink、阿里云AnalyticDB等。

其中，SQL生成器的上游和下游主要涉及：

• 上游计划中心


• 配置指标：用户在前端配置他想看的数据有哪些。
• 生产原始数据：根据用户输入得到哪些表作为数据源， 以及它们之间的连接关系。

• 下游Metric适配器


• 把SQL发布到Flink， 根据建表数据建物理表。

本文主要从技术角度介绍下SQL生成器相关的内容。

二 技术实现

在项目实施阶段，需要从需求分析、技术方案设计、测试联调几个步骤展开工作。本文重点不放在软件开发流程上， 而是就设计模式选择和数据结构算法设计做下重点讲解。

需求分析

在需求分析阶段， 我们明确了自动生成SQL模块所需要考虑的需求点， 主要包含如下几点：

• 需要支持多个事实表（流表）、多个维度表连表，其中一个事实表是主表，其他的均为辅助表。


• 维表变动也应当引起最终数据库更新。

• 主表对辅助表为1:1或N1，也就是说主表的粒度是最细的， 辅表通过唯一键来和主表连接。


• 流表中可能存在唯一键一致的多张流表， 需要通过全连接关联。唯一键不同的表之间通过左连接关联。

• 只有连表和UDF，没有groupby操作。


• 要求同步延时较小，支持多种源和目标介质。由于查询压力在目标介质，所以查询qps没有要求。

系统流程图

明确需求后， 我们把SQL生成器总体功能分为两块：

• 同步生成SQL和建表数据
• 异步发布SQL和建表

之所以把生成SQL阶段做成同步是因为同步阶段内存操作为主，如果发现数据有问题无法生成SQL能做到快速失败。发布阶段调用Metrics需要同步等待较长时间， 每个发布步骤要做到有状态记录， 可回滚或者重试。所以异步实现。SQL生成器同步阶段的整体功能细化到小模块，如下图所示：

检查阶段

检查原始数据是否有问题， 无法生成SQL则快速失败。

• 参数检查：检查上游是否提供了基本的参数， 比如事实表信息（可以没有维表， 但是必须有事实表）。


• 表类型检查：检查数据来源类型是否支持。

• 分区字段检查：是否提供了大宽表分区字段。


• 连接约束：检查流表，维表连接信息是否正确。

• 主表唯一性约束：检查主表是否含连接信息，唯一键是否有ETL信息。


• 元数据检查：检查是否包含HBase配置信息。

• 主键修正：修正维表连接键， 必须是维表的唯一键。

数据同步

• 同步所有原始表和原始表的连接数据（比如源表同步进来， 生成1:1的HBase表）。


• 生成优先级队列：生成连接和发布等任务的执行优先级。

• 同步填充：填充源表对应的同步阶段HBase表数据，和对应的配置项， 类型转换（比如源表是MySQL表，字段类型要转换为HBase的类型）， ETL填充， 添加消息队列（通过发送消息的方式通知下游节点运行）。


• 重复列修剪：删除重复的列。

• 空白列打标：对于满足一定条件（比如不需要在大宽表展示， 不是唯一键列， 连接键列， 保序列）的列打上空白列标识。


• 保序字段填充：如果上游提供了表示数据创建时间的字段， 则用该字段作为数据保序字段， 没有则填充系统接收到数据的时间作为保序字段。

计算阶段

生成大宽表，填充SQL。

• 中间表填充：填充全连接产生的中间表。


• 连接关系升级：会在本文后面说明。

• 反向索引填充：填充“反向索引”信息。


• 消息填充：中间表添加消息队列（中间表更新可以触发下游节点）。

• 大宽表填充：填充大宽表数据。


• 连接链对齐：中间表和大宽表连接键对齐。

• ETL填充：填充大宽表列的ETL信息。


• 分区字段填充：填充大宽表分区字段。

• SQL填充：填充Flink同步表映射SQL语句， Flink计算SQL语句， Flink结果表映射SQL语句。


• 保存：把SQL和建表数据存入数据库， 之后的请求可以复用已有的数据， 避免重复建表。

异步发布阶段会把SQL语句发布到Flink。

添加反向索引的原因

假如有A、B两表连接，那么连接方式为A表的非主键连接B表主键。从时序上来说可能有以下三种情况：

• B表数据先于A表数据多天产生
• B表数据后于A表数据多天产生
• B表数据和A表数据同时产生

下面我们就这三种情况逐一分析。

场景1：B表数据先于A表数据多天产生

我们假如B表数据存储于某个支持高qps的数据库内，我们可以直接让A表数据到来时直接连接此表（维表）来实现连表。

场景2：B表数据后于A表数据多天产生

这种场景比较麻烦。A表数据先行产生，因此过早的落库，导致B表数据到来时即使连接B维表也拿不到数据。这种场景还有一个类似的场景：如果AB连接完成后B发生了更新，如何让B的更新体现在宽表中？

为了解决这种问题，我们增加了一个“反向索引表”。假如A的主键是id，连接键是ext_id,那么我们可以将ext_id和id的值存储在一张表内，当B的数据更新时，用B的主键连接这种表的ext_id字段，拉取到所有的A表id字段，并将A表id字段重新流入Flink。

三 设计模式

对系统整体流程有了解以后， 我们再来看看系统的设计模式选择，选择设计模式时，我们考虑到数据处理相关的开发工作存在一些共性：

• 拆解后小功能多
• 小功能存在复用情况
• 小功能执行有严格的先后顺序
• 需要记录小功能运行状态， 流程执行可回滚或者中断可恢复执行

由于数据处理任务的步奏比较冗长，而且由于每个阶段的结果与下阶段的执行有关系，又不能分开。

参考 PipeLine（流水线）设计模式[2]，综合考虑后我们系统的整体设计如下图所示：

首先有一个全局的PipeLineContainer管理多个pipeLine和pipeline context， 每个pipeline可独立执行一个任务， 比如pipeline1执行同步生成sql任务。pipeline2执行异步发布任务。发布必须在生成SQL结束后执行， pipeline有状态并且按一定顺序串联。每个pipeline包含多个可重用的valve（功能）。valve可以重用， 任意组合，方便完成更多的数据处理任务（比如以后如果要支持Tisplus dump平台接入， 则简单拼接现有的valve就可以）。

四 数据结构和算法

问题说明

SQL生成器关键点， 就是把各个表（Meta节点）之间的关系表示出来。Meta之间的关系分为两类，分别是全连接关联和左连接关联（因为左连接关联涉及到数据的时序问题， 需要添加反向索引较为复杂， 所以和全连接区分了一下， 为了简化问题我们先执行全连接， 再执行左连接）。

我们要解决的问题是， 多个数据源同步数据进来之后， 按一定的优先级关联， 最终得到一个大宽表并需要自动发布。抽象到数据结构层面就是：

• 每个同步进来的数据源对应一个叶子节点
• 节点之间有关联关系，关联关系有多类并有执行优先级
• 所有节点和关联关系组成一棵树
• 最终得到一个根节点（大宽表）并发布

算法思路

下面说明下解决该问题的算法思路。

优先级队列

因为叶子节点之间连接执行优先级不同，先放入优先级队列。之后每次取出高优先级任务执行。相同优先级任务可以复用， 连续执行多次。优先级队列示意图如下：