【DTSE Tech Talk 精选问答】NO.55丨GaussDB(DWS)基于Flink的实时数仓构建

华为云数仓GaussDB(DWS)基于流处理框架Flink实现了高效实时数仓构建，数据分析时效从T+1时效趋向于T+0时效。GaussDB(DWS)+Flink如何增强湖仓增量数据在不同数据模型层之间的实时流动能力？如何为消息数据流提供高性能通用入库能力？本期直播将为您带来DWS-Flink的全新探索实践分享。
直播链接：https://bbs.huaweicloud.com/live/DTT_live/202404101630.html

Q：CN（可能指的是Catalog或某种配置组件）在Sink Connector中起到了什么作用？它如何影响数据的写入？

A:数据的写入是首先和cn建连，然后由cn再去将数据分发到各个dn。

Q：dws binlog的原理？

A:数据插入时会同步往一张辅助表中插入数据，该数据还额外记录了操作的类型。

Q：Flink SQL和GaussDB(DWS)函数有哪些区别？

A:2个不同的体系，可以分别看下各自官网对函数的介绍。

Q：Flink 有两种写法，SQL 和 API，GaussDB(DWS) 对 Flink SQL的支持怎么样？能否用SQL实现绝大部分需求？（毕竟对于企业来说，招一个会SQL的，远比一个会Flink API 的成本要低的多）另外，例如MySQL数据库和Fink SQL的语法还有所差异，不能直接用！GaussDB（DWS）是否考虑到这一点，通过代码处理掉兼容性问题？

A:当前一期，我们实现的正是SQL级的能力。分享中讲到的能力全部能用sql实现。 FlinkSQL有其特殊性，SQL解析是flink内核代码完成的。我们暂时没有考虑通过修改flink内核处理掉flinksql与其他sql的兼容性问题。

Q：Flink+GaussDB(DWS)的实时数据写入速度有哪些推荐优化思路？

A:调整攒批大小、攒批时间、写入方法。

Q：Flink的DataStream API是如何处理SQL查询中的JOIN操作的？

A:执行LookupTableSource的getLookupRuntimeProvider。

Q：Flink如何保证数据流处理的实时性？

A:通过实时读取binlog数据。

Q：Flink如何处理Cassandra中的数据？

A:自定义实现Cassandra source和sink。

Q：Flink如何定义和处理数据流？

A:flink中将流数据封装为DataStream，数据以DataStream的形式在各节点间传递。

Q：Flink如何知道在DWS中对应的schema是否存在？如果不存在，Flink会如何创建新的schema？

A:会和DWS建立连接（JDBC的方式）然后调用对于sql语句。

Q：Flink与GaussDB(DWS)对接时，在处理大量实时数据时如何避免数据倾斜问题，保持入库性能稳定？

A:数据倾斜需要观察作业中每个算子的处理数据量，查看到数据倾斜后，改变数据混洗的key来改变有倾斜的算子承接的数据量。

Q：Flink与GaussDB(DWS)在数据模型层之间流动数据时，数据格式的兼容性是如何保证的?怎么处理数据的乱序问题?

A:创建flink与dws数据类型对应关系。

Q：Flink中的DataStream API和DataSet API有什么区别？

A:前者是流处理后后者是批处理，不过现在可以统一使用前者。

Q：GaussDB(DWS)的查询计划和执行引擎在湖仓一体中如何提升查询性能？

A:针对hive/hudi bucket外表，dws优化了分布式计划的数据重分布逻辑。

Q：GaussDB(DWS)对Hive、Hudi、Iceberg等数据湖的兼容是如何实现的？

A:dws支持hivemetastore的方式读写hive，并且支持lakeformation的方式访问hive和hudi，Iceberg暂不支持。

Q：GaussDB(DWS)基于Flink的实时数仓构建的优势和挑战是什么？

A:flink作为实时计算引擎，dws作为存储、计算、支持海量数据离线处理和交互查询的数据仓库，flink+dws优势在于上游数据实时入库+数据入库后实时高效查询场景。

Q：GaussDB(DWS)如何提升入库性能以支持实时数据处理？

A:通过merge的方式写入Hstore表，支持批量插入，异步merge到表。

Q：GaussDB(DWS)如何支持ORC、Parquet等三方库的升级，并实现对AVRO格式的支持？

A:dws支持外表方式查询和写入orc和parquet数据。

Q：GaussDB(DWS)如何支持实时AP分析和高性能TP查询？

A:dws支持列存表和Hstore表，具有IO资源更少，读取性能好的优势。

Q：GaussDB(DWS)如何支持数据的高效点查？

A:dws支持主键等传统索引能力去重和加速点查，也支持分区、多维字典、局部排序等方式进一步加速AP查询。

Q：GaussDB(DWS)与Flink集成的具体机制是什么？在集成过程中，如何保证数据的一致性和可靠性？有哪些关键配置或参数对集成性能有显著影响？ 对于不同格式的消息数据流（如JSON、Avro、Protobuf等），GaussDB(DWS)如何提供高性能的通用入库能力？如何处理异构数据源的数据入库问题？有哪些机制可以保证数据入库的实时性和准确性？

A:我们提供了一个jar包，在配置flink任务时将对应的connector指定为dws即可；攒批的大小、写入的方式等都会入库的性能有影响（目前没有一个万能的配置，还是集体场景具体分析）。 对于入库来说，其实不用关注最上游数据是什么格式，因为它最终是从flink中读取的数据，即数据和定义的映射表是对应的。

Q：GaussDB(DWS)与Flink集成后，如何实现数据的实时流动和转换？例如，数据从Kafka等消息队列进入Flink后，如何高效地写入和更新GaussDB(DWS)中的数据模型？

A:如要将写入flink的数据同步到GaussDB(DWS)中，只需要集成我们connector的jar包，然后将sink端的connector指定为dws即可。

Q：GaussDB(DWS)与Flink结合时，如何管理内存资源以防止资源耗尽？

A:flink可以设置内存使用的上限。

Q：GaussDB(DWS)在实时数仓构建中与其他技术（如Hudi）的兼容性是如何体现的？

A:dws支持hudi作为外表操作，通过spark+avro的方式读取hudi文件数据后传到dws中。

Q：GaussDB(DWS)支持的操作场景有哪些，特别是在实时数仓构建中的应用案例？

A:dws列存表支持高速查询分析为主的AP场景，Hstore表支持实时并发入库场景。

Q：lf server1在湖仓一体架构中扮演什么角色？

A:dws当前正实现lakeformation的元数据对接，通过lakeformation可直接访问对应数据仓库，已支持hive、hudi等元数据管理。

Q：OBS在湖仓一体中如何存储和管理数据文件？

A:dws支持外表的方式读写obs数据。

Q：当数据量或复杂性增加时，如何保证系统的扩展性，以维持实时数据处理的能力？

A:GaussDB(DWS)集群和flink集群都可以横向扩容。

Q：当数据量增长或业务复杂度提升时，GaussDB(DWS)结合Flink的架构如何进行横向或纵向扩展？

A:对GaussDB(DWS)集群扩容，扩大flink集群的规模。

Q：对于不同格式的消息数据流（如JSON、Avro、Protobuf等），GaussDB(DWS)如何提供高性能的通用入库能力？

A:入库能力是基于flink的，即从flink中读取数据，无需关注写入flink之前的数据是何种格式。

Q：对于不同类型的消息数据流，GaussDB(DWS)是否提供了针对不同类型数据的适配器或者预处理功能来简化入库过程？

A:不需要，一个source算子可以承接多种数据类型。

Q：对于大规模的数据流，GaussDB(DWS)+Flink如何实现高效的负载均衡？如何通过Flink处理跨地域的数据流？

A:flink只对接kafka数据源，至于大规模跨地域数据流，应该解耦到kafka去解决，避免flink直面此问题。

Q：对于这三个连接器（Source Connector、Dimension Connector、Sink Connector），它们之间是如何协同工作的？

A:如果是普通读取会走source connector的逻辑；如果是维表join就会走dimension connector逻辑；如果是写入就会走sink connector逻辑。

Q：湖仓一体如何确保查询输出可以直接到仓内或数据湖，无需额外数据中转拷贝？

A:dws的sql引擎与存储引擎直接去湖中读取数据，天然解决了您提出的问题。

Q：湖仓一体如何实现无缝访问数据湖？

A:dws当前正实现lakeformation的元数据对接，通过lakeformation可直接访问对应数据仓库，已支持hive、hudi等元数据管理。

Q：华为云数仓GaussDB(DWS)基于流处理框架Flink，是如何解决数据快速出、入库的？

A:基于source算子，binlog算子和sink算子。

Q：老师可以详细介绍一下Dimension Connector吗？

A:在flink数据流处理过程中需要关联维度信息时，通过实现LookupTableSource接口的getLookupRuntimeProvider方法，flink会根据join条件查询维表数据后与数据流关联。

Q：批量计算和增量计算在GaussDB(DWS)中有何不同？

A:前者可以理解为只处理一次；后者是随着数据的到来不断的处理。

Q：请问随着业务的发展和数据量的增长，如何确保GaussDB(DWS) + Flink方案的可扩展性？同时，在面临网络中断、硬件故障等异常情况时，如何确保方案的容错性和高可用性？

A:GaussDB(DWS)集群和flink集群都可以横向扩容；主要通过flink的checkpoint机制保障。

Q：如果DWS的元数据发生变化（如添加新表或修改表结构），Flink如何同步这些变化？

A:Schema Evolution 设计，自动获取schema变更。

Q：如果flink copy的时间超过积攒数据的时间，如何避免kafka积压？

A:优化sink写入性能，增加并发，提升资源。

Q：如果Flink任务因发生故障而停止，恢复缺失数据的方案是怎样的呢？

A:通过checkpoint或savepoint恢复。

Q：如果一个Sink Connector需要写入多个DN，它是如何处理这种情况的？

A:sink connector目前只能连接cn，由cn去分发要写的dn。

Q：如何从CN中读取DN（可能是指数据节点或目标存储的分区）的数量和分布列？

A:可以通过pgxc_node视图来查看；分布列可以通过\d+ 表名来查看。

Q：如何定义和使用GaussDB(DWS)中的维度（Dimension）？

A:如果flink中有维表join的任务，就会从维表中读取匹配到的数据。

Q：如何将复杂的SQL下推到GaussDB(DWS)中进行处理？

A:目前只会将limit语句、projection投影以及where查询条件下推到GaussDB(DWS)。

Q：如何配置Flink作业以最佳地利用GaussDB(DWS)的分布式存储和计算能力？

A:需要根据dws的集群规模，增量数据量，性能要求等综合设计与测试。

Q：如何实现Flink作业与GaussDB(DWS)表结构变更的自动化同步？

A:Schema Evolution 设计，自动获取schema变更。

Q：如何使用Flink高效地捕捉和处理GaussDB(DWS)中的增量数据？

A:通过读取binlog数据来实现。

Q：如何通过HStore Binlog表实现实时数据导出？

A:集成我们的connector jar包，然后新建flink任务时将connector指定为dws，并将binlog属性指定为true即可。

Q：如何通过SELECT语句从ex3.T1和internal schema.T2中联合查询数据？

A:dws支持external schema表和internal表的联合查询。

Q：如何统一批流处理逻辑以简化数据生产线架构的复杂度？

A:Source算子，binlog算子天然支持批流一体，无需额外设计。

Q：如何有效地更新GaussDB(DWS)中的维度表（DIM层）？

A:在GaussDB(DWS)中维度表的更新和普通表的更新没区别；flink中可以设置对应维度表的缓存时间和大小。

Q：若copy的时间超过积攒数据的时间，如何避免kafka积压？

A:避免kafka积压就要加大flink的并行度。

Q：实时数仓是否支持postgresql的驱动？

A:不建议使用postgresql驱动，可以使用我们提供的驱动。

Q：实时数据处理中，GaussDB(DWS)+Flink如何处理数据的并发读写？如何保障数据的实时分发和调度？

A:对于流读的话，flink会以dn为粒度来启动任务，即最好的场景是flink任务数和集群中的dn数相等；对于写的话，默认是按照主键来分发到不同的任务中，所以最要使用upsert的方式来写入。

Q：使用Flink-CDC迁移要对迁入组件一般需要对哪些部分代码重写？

A:迁入和迁出分别需要自定义实现DataSinkFactory和DataSourceFactory接口。

Q：要增加GaussDB的CDC，是否需要对每种数据库的flink-connector重写？

A:只需要增加gaussDB的source connector即可。

Q：遇到"No Route to Host"错误时，如何进行故障排查与恢复？

A:判断网络是否连通或集群是否授权。

Q：在实时数仓中，如何确保数据的准确性和一致性，特别是在处理高并发写入时？

A:一般使用upsert的方式入库，可以保证幂等性。

Q：在实现T+0时效的目标时，Flink对实时流数据的处理策略是怎样的？

A:实时读取binlog数据。

Q：在使用GaussDB(DWS)和Flink进行湖仓增量数据实时流动时，如何保证数据的准确性和一致性？在数据同步过程中，如果出现异常或延迟，有哪些机制可以确保数据不丢失或重复？

A:通过checkpoint机制能保证at least once语义。

Q：在未来的发展中，Flink与DWS的元数据对接有哪些潜在的改进和优化方向？

A:目前flink和dws的元数据是相互打通的，有序可能和其他的组件进行打通。

Q：增量计算方案图是如何由Binlog流驱动整个任务的？

A:flink任务会实时从数据库中获取binlog数据，如果获取到了就往下游推送，下游的sink任务就会接受到对应的数据然后将数据写入到另外的表中。

Q：增量计算能力如何支持复杂事件处理（CEP）？

A:CEP是flink的能力之一，可直接使用之，source算子使用分享中讲到的算子即可。

Q：增量计算在GaussDB(DWS)中是如何实现的？

A:flink通过实时读取binlog信息，然后执行增量计算的逻辑。

Q：智能湖仓流对数据格式有哪些要求?

A:仓内是私有格式，湖中是开源格式。

Q：自定义实现的Flink CustomPartitionerWrapper与Flink内置的Partitioner有何不同？

A:自定义的Partitioner规则是和cn分发数据到dn的规则是一样的。

想要了解更多相关知识，欢迎观看DTSE Tech Talk 系列技术直播