打造 Flink + StarRocks+ Dinky 的极速统一分析平台

摘要：本文介绍了打造 Flink + StarRocks + Dinky 的极速统一分析平台经验分享。内容包括：

1. 背景
2. 技术架构
3. 应用实践
4. 开发运维改善
5. 总结
6. 未来规划

Tips：历史传送门～

《Dinky 开源一周年了~》

《Dinky 扩展 ChunJun 的实践分享》

《Dinky 扩展 kudu 实践分享》

《Dinky 构建 Flink CDC 整库入仓入湖》

GitHub 地址

https://github.com/DataLinkDC/dlink

https://gitee.com/DataLinkDC/Dinky

欢迎大家关注 Dinky 的发展~

一、背景

青花瓷集团是一家以传统养生为主体的服务性公司。致力于打造从传统模式到数字化转型的线上模式。

随着公司业务的快速发展，为满足业务团队实时报表统计和决策分析，我司选择基于 Apache Flink + Starocks + Dinky 构建的极速统一分析平台。

二、技术架构

目前采用 Lambda 架构，实现实时数据和离线数据相结合，统一数据到 Starrocks 做进一步的数据分析。架构如下：

离线计算

采用 sqoop 将 mysql 数据按照 T+1 的方式，每天加载到 Hive 做一些维表字段的冗余；另外一些 Mysql 全量数据和 Hive 通过 Flink Batch 同步到 Starrocks。其中 Mysql 全量跑批是通过 Flink Batch 5 分钟跑批(涉及到特殊场景的表)。

实时计算

MySQL 业务数据部分采用 Dinky 整库同步全量 + 增量的方式同步，部分采用Canal + Kafka + Flink 增量和 Starrocks MySQL 外部表全量的方式同步，以达到实时更新的目的，写入Starrocks 的主键模型表；行为日志通过 FileBeat + Kafka + Flink 的方式写入 Starrocks 的明细表。

统一数据分析平台

Dinky 提供了 Flink 上的批处理和流计算能力，以及外部数据库查询与操作的能力，使得我们的开发效率进一步提升。此外还可以直接在 Dinky 进行数据分析和 ETL 处理，避免了在服务器上部署各种脚本。

三、应用实践

数据模型选择

Starocks 的数据模型表是一种以 key-value 键值对存在的列式存储。当前支持的模型有明细模型（Duplicate Key）、聚合模型（Aggregate Key）、更新模型（Unique Key）和主键模型（Primary Key）。数据模型的应用场景可以参考 Starrocks 官网的数据模型介绍。根据我司业务报表需求对 Delete，Update 操作比较频繁，其次对于商城系统的行为数据要求实时同步为用户画像做沉淀。因此目前采用了主键模型（Primary Key）和明细模型（Duplicate Key）做支撑。

整库同步

对于数据同步，初期调研的时候有3种方案：

• Flink CDC
• Canal + Kafka + Flink
• Canal + Kafka + Routine Load

下面分别说下 3 种方案各有优缺点

Flink CDC

优点：支持全量和增量，并且支持断点续传和 ETL，目前支持的数据源也越来越丰富；

缺点：一张表对应一个 JDBC事务，如果连接数过多，容易对业务库造成压力。不支持整库同步。

Canal + Kafka + Flink

优点：当时考虑到业界的一种通用方案 ；

缺点：只支持增量，全量数据需要另外脚本实现。

Canal + Kafka + Routine Load

优点：简化额外组件，可以很方便在 Starrocks 做数据同步；

缺点：在 Starrocks 2.1 之前的版本对主键模型支持不完善。

基于以上几种方案在前期从 Flink CDC 到 Canal + Kafka + Flink 再到 Canal + Kafka + Routine Load的不同程度迁移，出现了很多坑。目前 Dinky 社区开发了基于 Flink CDC 整库同步的功能后，经过多方面和社区的沟通，Flink CDC 整库同步已在线上平滑迁移运行，也极大降低了对业务库的压力。

Dinky 整库同步介绍

Dinky 定义了 CDCSOURCE 整库同步的语法，该语法和 CDAS 作用相似，可以直接自动构建一个整库入仓入湖的实时任务，并且对 source 进行了合并，不会产生额外的 Mysql 及网络压力，支持对任意 sink 的同步，如 kafka、doris、hudi、jdbc 等。其原理是采用只构建一个 source，然后根据 schema、database、table 进行分流处理，分别 sink 到对应的表。

Dinky 整库同步语法

EXECUTE CDCSOURCE jobname WITH (
  'connector' = 'mysql-cdc',
  'hostname' = '192.168.0.2',
  'port' = '3306',
  'username' = 'root',
  'password' = '123456',
  'checkpoint' = '3000',
  'scan.startup.mode' = 'initial',
  'parallelism' = '1',
  'table-name' = 'test\.student,test\.score',
  'sink.connector' = 'starrocks',
  'sink.jdbc-url' = 'jdbc:mysql://192.168.0.3:19035',
  'sink.load-url' = '192.168.0.3:18035',
  'sink.username' = 'root',
  'sink.password' = '123456',
  'sink.sink.db' = 'qhc_ods',
  'sink.table.prefix' = 'ods_bak_',
  'sink.table.lower' = 'true',
  'sink.database-name' = 'qhc_ods',
  'sink.table-name' = '${tableName}',
  'sink.sink.properties.format' = 'json',
  'sink.sink.properties.strip_outer_array' = 'true',
  'sink.sink.max-retries' = '10',
  'sink.sink.buffer-flush.interval-ms' = '15000',
  'sink.sink.parallelism' = '1'
)

Flink 提交模式

在统一数据分析平台 Dinky 中已经支持的模式包括 StandAlone、Yarn Per Job、Yarn Application、Yarn Session、K8S Session、K8S Application。在项目初期由于各种原因，先选择了 StandAlone 在 Dinky 提交 FlinkSQL 任务。随着业务的稳定，现在已经将绝大多业务迁移到 Yarn Application 之上。

在 Yarn Application 测试过程当中，也出现了一个比较重要的问题，当 Yarn 是高可用时，提交 Yarn Application 会出现作业重复的问题。经过和 Dinky 社区不断的沟通，最终解决了 Yarn Application 在高可用情况下，作业重复的问题。在这里要非常感谢 Dinky 社区，问题反馈后，也很快得到了解决。

如果要部署 Yarn Application 模式，首先需要将 FLINK_HOME/lib 下的包上传到 HDFS。其次需要将 dlink-app-0.6.x.jar 上传到和 FLINK_HOME/lib 包对应的 HDFS 目录。

上传完成后，需要给与这些包对应的权限。

#每个包都给与这个权限
hadoop fs -chmod o+rx hdfs://nameservice1/flink_yarn/lib/

最后，需要在 Dinky 修改提交 FlinkSQL 的 Jar 文件路径。如下：

需要说明的是，由于目前 Dinky 整库同步在 Yarn 上只支持 Session 模式与 Per Job 模式。在加上我司既有 Flink 同步任务也有 ETL 任务，所以 Yarn Session 和 Yarn Application 模式两者都有。待 Dinky 整库同步支持 Application 后，在做进一步迁移。

外部表统一分析

Starocks 除自身的几种数据模型外，还提供了对外部数据源的支持，如 Mysql、Hive、ElasticSearch、Hudi 等。

目前离线计算维度表通过 Hive 外部表，每天凌晨跑一次全量。对于第三方数据是存在与Mysql 业务库，因一些特殊原因，无法开启 binlog，只能通过准实时的方式，每2小时跑一次批。

目前所有的外部表都是通过在 Dinky 之上做同步，极大的降低了开发成本。需要注意的一点是目前 Dinky 只支持部分 Starocks 的语法，如INSERT，TRUNCATE等。对于 CREATE EXTERNAL 语法创建外部表还不支持。因此 DDL 语法在 MySQL客户端执行，INSERT，TRUNCATE 在 Dinky 执行。

Hive外部表

-- 创建一个名为 hive0 的 Hive 资源
CREATE EXTERNAL RESOURCE "hive0"
PROPERTIES (
  "type" = "hive",
  "hive.metastore.uris" = "thrift://bigdata1:9083"
);

-- hive外部表
CREATE EXTERNAL TABLE `sta_dim_employee_dwd` (
  `id` bigint(20) NULL COMMENT "",
  `performance_id` bigint(20) NULL COMMENT "",
  `goods_type` int(11) NULL COMMENT "",
  `goods_type_refine` int(11) NULL COMMENT "",
  `goods_specs` varchar(65533) NULL COMMENT ""
) ENGINE=HIVE 
COMMENT "PARTITION BY ()"
PROPERTIES (
"database" = "dws",
"table" = "dim_employee_dwd",
"resource" ="hive0",
"hive.metastore.uris" = "thrift://bigdata1:9083,thrift://bigdata2:9083"
);

Mysql外部表

CREATE EXTERNAL TABLE `sta_assassin_employee` (
  `id` largeint(40) NOT NULL COMMENT "主键",
  `old_id` varchar(65533) NULL COMMENT "",
  `tenant_id` varchar(65533) NULL COMMENT "租户ID",
  `leader_id` largeint(40) NULL COMMENT "直属上级",
  `code` varchar(65533) NULL COMMENT "职员工号"
) ENGINE=MYSQL 
COMMENT "MYSQL"
PROPERTIES (
"host" = "127.0.0.1",
"port" = "3306",
"user" = "root",
"password" = "123456",
"database" = "employee",
"table" = "assassin_employee"
);

Dinky Starocks 作业

作业 ETL 调度

当前 Dinky 还不支持完善的调度，如果遇到批处理的情况。Dlinky 目前提供了通过 Openapi的方式结合第三方调度平台做批作业的周期调度，前提是第三方调度平台支持 http。Dinky Openapi 的示例如下:

说明: 其中的 id 是所在 dinky 元数据库 dlink_task 表中对应的 id。

四、开发运维改善

在大数据出现后，业界并没有什么可用的开发工具提升开发效率，初期的时候大家都是用编程语言写业务逻辑额外增加了开发测试周期。随着大数据的这几年的不断发展，大数据逐步的向 SQL 化发展。虽然一定程度上降低了开发测试周期，但由于大数据组件解决不同的应用场景，往往使得开发测试需要在不同的平台。

由于以上原因，Dataops 应用而生。Dataops 架构解决什么问题，什么场景，如何解决的。这里就不叙述了。感兴趣的可以查看相关资料。这种架构我所了解的是阿里云算是一个先行者，极大的解决了各种组件在统一开发平台处理的能力。随着这种架构被业界所熟知，各公司也开始研发这种类似的平台。开源方面现在也涌现了很多类似平台，诸如 Streamx，Dinky 以及微众开源的 DSS。

那我们为什么选择了 Dinky，而不是 Streamx 或者微众开源的 DSS呢？不是说 Streamx 或者微众开源的 DSS 不好。工具而言，都各有各的优势和使用场景。首先，Dinky 是基于 Flink之上的数据开发平台，方便我们采用 FlinkSQL 做实时同步和实时 ETL；其次是 Dinky 提供了一站式的能力，在开发效率、运维上都极大的降低了我们的开发成本。还有一点就是 Dinky 的未来的发展方向上也更符合我们的场景需求。基于以上几点，我们选择了 Dinky 做为我们的数据开发统一平台。对于使用 Dinky 前和使用 Dinky 后的改善，主要罗列如下几点：

五、总结

综上，通过 Flink + Starrocks + Dinky 构建了一套数据统一分析平台，这套分析平台让我们能够进行高效实时的数据分析。同时在 Flink CDC 整库同步 和 Yarn Application 模式都为我们开发解决了很多问题。

我们也一直在跟进 Dinky 社区的发展，初期在使用过程当中也遇到了很多问题。非常感谢社区小伙伴能够及时帮忙解决问题。

在使用过程中，我们发现了几点问题：

1.在当前版本中，租户、权限等还不支持，好在社区在开发中；

2.调度不完善，社区已经在开发中。

六、未来规划

1.Dinky 批处理调度未来采用 Apache DolphinScheduler；

2.租户、权限完善后，做进一步版本迭代。