一种基于Flink数据流的Iceberg表Schema动态变更方法及系统

摘要

本发明公开了一种基于Flink数据流的Iceberg表Schema动态变更方法，创建数据行，并组装成数据流，其中所述数据流为附带结构信息的数据流；实现Flink算子IcebergStreamWriter，通过该算子接收所述步骤1中组装的数据流记录，并存储到数据文件中，在存储到数据文件之前检测接收到的数据是否存在结构信息；实现Flink算子IcebergFilesCommitter，通过该算子将步骤2中生成的数据文件组装成清单文件，并生成快照。相比现有的实现方案能够做到自动映射数据行的结构信息到目标表结构，不必再重新修改代码，重启作业，同时数据行记录并不是每次都需要携结构信息，只有在第一条记录和结构发生变更时才需要，因此相比原有方式，并不会带来网络和计算开销，此外该方案对原有方案的侵入性非常小，实现起来也很简单。

说明书

技术领域

本发明涉及大数据实时流计算领域，尤其涉及一种基于Flink数据流的Iceberg表Schema动态变更方法及系统。

背景技术

Iceberg是目前最流行的数据湖技术框架之一，Iceberg作为HDFS、S3等存储系统之上的抽象层，提供了一种独立的数据组织方式，可实现大数据场景下的数据的追加、修改和删除，表Schema的变更等，上述功能以事务快照的形式实现，支持表的并发读写、数据回溯和版本回滚等。Iceberg和Flink结合，能够实现实时数据入湖和实时数仓等功能。

Iceberg本身支持表Schema的动态变更，但是跟Flink结合时，因为Flink的设计实现等原因，导致在创建数据流时需要明确指定数据行DataRow的Schema，即数据行的数据结构，而且一旦指定数据行的数据结构信息，该结构信息就无法在作业流运行过程中动态修改，必须人工停止作业再重新修改数据结构和Iceberg目标表的结构，然后再重启作业。在实际应用中，数据行的数据结构发生变更是常有的事，因此通过修改代码再重启作业，在实时应用场景是无法接受的，出现该问题的根源是数据行DataRow本身不带有Schema信息，因此本发明专利通过在数据行中引入结构信息，在作业运行过程中根据结构信息的变更自动映射到目标表结构，以此来解决必须重启作业的问题。

发明内容

本发明旨在至少解决现有技术中存在的技术问题之一。为此，本发明公开一种基于Flink数据流的Iceberg表Schema动态变更方法，所述方法包括：

步骤1，创建数据行，并组装成数据流，其中所述数据流为附带结构信息的数据流；

步骤2，实现Flink算子IcebergStreamWriter，通过该算子接收所述步骤1中组装的数据流记录，并存储到数据文件中，在存储到数据文件之前检测接收到的数据是否存在结构信息；

步骤3，实现Flink算子IcebergFilesCommitter，通过该算子将步骤2中生成的数据文件组装成清单文件，并生成快照。

更进一步地，所述组装成数据流进一步包括：

步骤101，如果数据行是第一条记录，则必须附带上数据行的结构信息；

步骤102，如果数据行不是第一条记录，且当前数据行的结构信息未变更，则当前数据行可以不必附带结构信息；

步骤103，如果数据行不是第一条记录，且当前数据行的结构信息发生变更，则当前数据行必须附带结构信息。

更进一步地，所述步骤2中的存储到数据文件中，在写入文件之前执行如下步骤：

步骤201，如果当前接收到的是第一条数据记录，检查是否存在结构信息，如果没有，则报告异常，作业终止；

步骤202，如果当前接收到的是第一条数据记录，检查是否存在结构信息，如果有结构信息，则根据提供的结构信息，创建Iceberg目标表和存储文件。

更进一步地，所述步骤2中的存储到数据文件中，在写入文件之前执行如下步骤：步骤203，如果当前接收到的不是第一条数据记录，检查当前行是否发送结构变更，如果结构信息发生变更，且提供了新的结构信息，则根据新提供的结构信息，修改Iceberg目标表结构，并创建新的存储文件。

更进一步地，所述步骤2中的存储到数据文件中，在写入文件之前执行如下步骤：

步骤204，如果当前接收到的不是第一条数据记录，检查当前行是否发送结构变更，如果结构信息发生变更，且没有提供新的结构信息，则报错，并终止作业；

步骤205，根据最新的表结构写入数据行记录。

本发明还公开了一种基于Flink数据流的Iceberg表Schema动态变更系统，所述系统包括：

数据流组装模块，创建数据行，并组装成数据流，其中所述数据流为附带结构信息的数据流，如果数据行是第一条记录，则必须附带上数据行的结构信息；如果数据行不是第一条记录，且当前数据行的结构信息未变更，则当前数据行可以不必附带结构信息；如果数据行不是第一条记录，且当前数据行的结构信息发生变更，则当前数据行必须附带结构信息；

数据存储模块，实现Flink算子IcebergStreamWriter，通过该算子接收所述数据流组装模块中组装的数据流记录，并存储到数据文件中，对接收到的数据进行检测，如果当前接收到的是第一条数据记录，检查是否存在结构信息，如果没有，则报告异常，作业终止，如果当前接收到的是第一条数据记录，检查是否存在结构信息，如果有结构信息，则根据提供的结构信息，创建Iceberg目标表和存储文件；

快照生成模块，实现Flink算子IcebergFilesCommitter，通过该算子将数据存储模块中存储的数据文件组装成清单文件，并生成快照。

更近一步地，所述数据存储模块进一步包括：如果当前接收到的不是第一条数据记录，检查当前行是否发送结构变更，如果结构信息发生变更，且提供了新的结构信息，则根据新提供的结构信息，修改Iceberg目标表结构，并创建新的存储文件。

更进一步地，如果当前接收到的不是第一条数据记录，检查当前行是否发送结构变更，如果结构信息发生变更，且没有提供新的结构信息，则报错，并终止作业；根据最新的表结构写入数据行记录。

本发明进一步公开了一种基于Flink数据流的Iceberg表Schema动态变更的装置，包括至少一个存储器和至少一个处理器；所述至少一个存储器，用于存储机器可读程序；所述至少一个处理器，用于调用所述机器可读程序，执行上述的变更方法的步骤。

更进一步公开了一种计算机可读存储介质，存储有能够被处理器加载并执行上述方法的计算机程序。

本发明于现有技术相比，有益效果为：本发明专利相比现有的实现方案能够做到自动映射数据行的结构信息到目标表结构，不必再重新修改代码，重启作业，同时数据行记录并不是每次都需要携结构信息，只有在第一条记录和结构发生变更时才需要，因此相比原有方式，并不会带来网络和计算开销，此外该方案对原有方案的侵入性非常小，实现起来也很简单。

附图说明

从以下结合附图的描述可以进一步理解本发明。图中的部件不一定按比例绘制，而是将重点放在示出实施例的原理上。在图中，在不同的视图中，相同的附图标记指定对应的部分。

图1是本发明的基于Flink数据流的Iceberg表Schema动态变更方法的流程图；

图2是本发明一实施例中动态变更的流程框图。

具体实施方式

实施例一

如图1所示，本实施例总体流程分为3个阶段：

第一阶段，创建数据行，并组装成数据流，组装数据流按照如下方式：

A.如果数据行是第一条记录，则必须附带上数据行的结构信息；

B.如果数据行不是第一条记录，且当前数据行的结构信息未变更，则当前数据行可以不必附带结构信息；

C.如果数据行不是第一条记录，且当前数据行的结构信息发生变更，则当前数据行必须附带结构信息；

第二阶段，实现Flink算子IcebergStreamWriter，该算子作用是接收阶段1中组装的数据流记录，并存储到数据文件中，写入文件之前需要按照如下顺序做以下事情：

如果当前接收到的是第一条数据记录，检查是否存在结构信息，如果没有，则报告异常，作业终止；

如果当前接收到的是第一条数据记录，检查是否存在结构信息，如果有结构信息，则根据提供的结构信息，创建Iceberg目标表和存储文件；

如果当前接收到的不是第一条数据记录，检查当前行是否发送结构变更，如果结构信息发生变更，且提供了新的结构信息，则根据新提供的结构信息，修改Iceberg目标表结构，并创建新的存储文件；

如果当前接收到的不是第一条数据记录，检查当前行是否发送结构变更，如果结构信息发生变更，且没有提供新的结构信息，则报错，终止作业；

根据最新的表结构写入数据行记录

3.实现Flink算子IcebergFilesCommitter，该算子作用是将步骤2中生成的数据文件组装成清单文件，并生成快照。

实施例二

从硬件角度描述本发明的发明构思，本实施例还公开了一种基于Flink数据流的Iceberg表Schema动态变更系统，所述系统包括：

数据流组装模块，创建数据行，并组装成数据流，其中所述数据流为附带结构信息的数据流，如果数据行是第一条记录，则必须附带上数据行的结构信息；如果数据行不是第一条记录，且当前数据行的结构信息未变更，则当前数据行可以不必附带结构信息；如果数据行不是第一条记录，且当前数据行的结构信息发生变更，则当前数据行必须附带结构信息；

数据存储模块，实现Flink算子IcebergStreamWriter，通过该算子接收所述数据流组装模块中组装的数据流记录，并存储到数据文件中，对接收到的数据进行检测，如果当前接收到的是第一条数据记录，检查是否存在结构信息，如果没有，则报告异常，作业终止，如果当前接收到的是第一条数据记录，检查是否存在结构信息，如果有结构信息，则根据提供的结构信息，创建Iceberg目标表和存储文件；

快照生成模块，实现Flink算子IcebergFilesCommitter，通过该算子将数据存储模块中存储的数据文件组装成清单文件，并生成快照。

更近一步地，所述数据存储模块进一步包括：如果当前接收到的不是第一条数据记录，检查当前行是否发送结构变更，如果结构信息发生变更，且提供了新的结构信息，则根据新提供的结构信息，修改Iceberg目标表结构，并创建新的存储文件。

更进一步地，如果当前接收到的不是第一条数据记录，检查当前行是否发送结构变更，如果结构信息发生变更，且没有提供新的结构信息，则报错，并终止作业；根据最新的表结构写入数据行记录。

本发明进一步公开了一种基于Flink数据流的Iceberg表Schema动态变更的装置，包括至少一个存储器和至少一个处理器；所述至少一个存储器，用于存储机器可读程序；所述至少一个处理器，用于调用所述机器可读程序，执行上述的变更方法的步骤。

更进一步公开了一种计算机可读存储介质，存储有能够被处理器加载并执行上述方法的计算机程序

如图2所示，数据文件的写和数据文件的元数据提交，分别通过Flink的两个算子IcebergStreamWriter和IcebergFilesCommitter来实现：

其中，算子IcebergStreamWriter用来写入记录到对应的avro、parquet、orc文件；IcebergStreamWriter借助Flink DataStream的并发机制，将数据文件的写操作并行化，可互不影响地写数据；IcebergFilesCommitter收集；

算子IcebergStreamWriter创建的数据文件，来创建manifest数据清单文件和快照的提交。

还需要说明的是，术语“包括”、“包含”或者其任何其他变体意在涵盖非排他性的包含，从而使得包括一系列要素的过程、方法、商品或者设备不仅包括那些要素，而且还包括没有明确列出的其他要素，或者是还包括为这种过程、方法、商品或者设备所固有的要素。在没有更多限制的情况下，由语句“包括一个……”限定的要素，并不排除在包括所述要素的过程、方法、商品或者设备中还存在另外的相同要素。

本领域技术人员应明白，本申请的实施例可提供为方法、系统或计算机程序产品。因此，本申请可采用完全硬件实施例、完全软件实施例或结合软件和硬件方面的实施例的形式。而且，本申请可采用在一个或多个其中包含有计算机可用程序代码的计算机可用存储介质(包括但不限于磁盘存储器、CD-ROM、光学存储器等)上实施的计算机程序产品的形式。

虽然上面已经参考各种实施例描述了本发明，但是应当理解，在不脱离本发明的范围的情况下，可以进行许多改变和修改。因此，其旨在上述详细描述被认为是例示性的而非限制性的，并且应当理解，以下权利要求(包括所有等同物)旨在限定本发明的精神和范围。以上这些实施例应理解为仅用于说明本发明而不用于限制本发明的保护范围。在阅读了本发明的记载的内容之后，技术人员可以对本发明作各种改动或修改，这些等效变化和修饰同样落入本发明权利要求所限定的范围。