SeaTunnel Assert Sink 连接器完全指南用规则化校验构建可信赖的数据管道【免费下载链接】seatunnelSeaTunnel is a multimodal, high-performance, distributed, massive data integration tool.项目地址: https://gitcode.com/GitHub_Trending/se/seatunnel本指南围绕 Apache SeaTunnel 的 Assert Sink 连接器展开。Assert 是一个零外部依赖的终端校验型 Sink它不向任何外部系统写数据而是按照用户在配置中声明的规则对管道输出进行断言行数是否落在期望区间、字段类型与取值是否合法、字段是否可空、多表场景下各表名与元数据是否与预期一致。读完本文你将掌握 Assert 全部规则体系field_rules/row_rules/catalog_table_rule/tables_configs的配置方法与流式/批式语义并能结合源码理解每条规则的执行时机与失败行为从而把数据质量校验直接内嵌到 SeaTunnel 作业中。一、Assert 是什么一个没有外部系统的终端 SinkAssert 是 SeaTunnel 连接器体系中的校验型 Sink 连接器其定位与 Kafka、JDBC 等写入型 Sink 完全不同它没有外部系统可写唯一的作用是接住上游数据并对它们做校验。官方文档的描述是它通过用户自定义规则检查行数row count、字段类型field type、字段值field value以及 Catalog 表元数据catalog table metadata一旦实际数据与规则不匹配作业就会失败job fails。这个定位让 Assert 在以下场景中极具价值管道自检Pipeline Self-check在开发或回归测试阶段把 Assert 挂在管道末端验证 FakeSource 或任意上游产出的数据是否符合预期无需准备下游数据库中间结果验证配合 Transform 使用校验转换后的中间结果避免脏数据流向下游多表作业校验对一个作业中的多张输入表分别断言行数与字段规则流式作业的累积校验在流式模式下对 Sink Writer 生命周期内累计接收的行数做最终校验。从源码结构看Assert 连接器由以下几部分构成seatunnel-connectors-v2/connector-assertsink/AssertSink、AssertSinkFactory、AssertSinkOptions选项定义、AssertSinkWriter校验执行入口rule/AssertFieldRule字段规则模型与规则类型枚举、AssertRuleParserHOCON 配置解析、AssertTableRule、AssertCatalogTableRuleCatalog 元数据规则excecutor/AssertExecutor单条数据校验执行器被AssertSinkWriter复用exception/AssertConnectorErrorCode、AssertConnectorException统一异常。二、引擎支持与功能特性支持的引擎Assert 连接器同时支持三大执行引擎Spark / Flink / Seatunnel Zeta功能特性一览特性支持exactly-once否cdc否batch✅stream✅多表写入multiple table write✅timer flush否需要说明的是Assert 是终端校验型 Sink本身不涉及恰好一次之类的语义保证它也不将UPDATE/DELETE行类型解释为 CDC 操作而是对接收到的每一行都执行规则断言见原文档 sink/Assert.md 中的 tip 说明。三、选项总览整棵规则树Assert 连接器只有顶层一个必填参数rules其余全部为嵌套可选规则。完整选项表如下字段名与 AssertConfig.java 中定义的常量一一对应名称类型必填默认值rulesConfigMapyes-rules.field_rulesConfigListno-rules.field_rules.field_namestring|ConfigMapyes-rules.field_rules.field_typestringno-rules.field_rules.field_valueConfigListno-rules.field_rules.field_value.rule_typestringno-rules.field_rules.field_value.rule_valuenumericno-rules.field_rules.field_value.equals_toboolean|numeric|string|ConfigList|ConfigMapno-rules.row_rulesConfigListno-rules.row_rules.rule_typestringno-rules.row_rules.rule_valuestringno-rules.catalog_table_ruleConfigMapno-rules.catalog_table_rule.primary_key_ruleConfigMapno-rules.catalog_table_rule.primary_key_rule.primary_key_namestringno-rules.catalog_table_rule.primary_key_rule.primary_key_columnsConfigListno-rules.catalog_table_rule.constraint_key_ruleConfigListno-rules.catalog_table_rule.constraint_key_rule.constraint_key_namestringno-rules.catalog_table_rule.constraint_key_rule.constraint_key_typestringno-rules.catalog_table_rule.constraint_key_rule.constraint_key_columnsConfigListno-rules.catalog_table_rule.constraint_key_rule.constraint_key_columns.constraint_key_column_namestringno-rules.catalog_table_rule.constraint_key_rule.constraint_key_columns.constraint_key_sort_typestringno-rules.catalog_table_rule.column_ruleConfigListno-rules.catalog_table_rule.column_rule.namestringno-rules.catalog_table_rule.column_rule.typestringno-rules.catalog_table_rule.column_rule.column_lengthintno-rules.catalog_table_rule.column_rule.nullablebooleanno-rules.catalog_table_rule.column_rule.default_valuestringno-rules.catalog_table_rule.column_rule.commentcommentno-rules.table-namesConfigListno-rules.tables_configsConfigListno-rules.tables_configs.table_pathStringno-multi_table_sink_replicaintno-common-optionsno-重要约束虽然只有rules是必填项但嵌套的规则块是可选的至少要配置一条有意义的规则否则 Sink 将无内容可校验。从源码看顶层rules选项的定义位于 AssertSinkOptions.java其类型为MapString, Object无默认值、必填而AssertSinkOptions本身继承自SinkConnectorCommonOptions因此multi_table_sink_replica与 common options 均来自公共 Sink 选项体系。四、核心规则详解4.1 rulesConfigMap必填规则的唯一必填顶层块描述期望数据。每个规则代表一种校验字段校验、行数校验、表名校验或 Catalog 表校验。4.2 field_rulesConfigList—— 字段级校验当需要校验字段类型、空值约束、值域范围、字符串长度或精确取值时使用。每条 field rule 由三部分构成field_namestring必填被校验的字段名。从 AssertExecutor.java 的实现看执行时会先在SeaTunnelRowType中按字段名查找索引如果字段不存在会直接抛出IllegalArgumentExceptionField name %s not found in row type %sfield_typestring | ConfigMap可选字段类型声明声明方式需遵循 schema-feature 文档 中如何声明受支持的类型一节的约定。例如string、int、decimal(30, 8)、arrayint、maptime, string嵌套行类型则直接以 ConfigMap 形式声明field_valueConfigList可选字段值的校验规则列表可叠加多条规则例如同时校验非空与长度上下限。4.3 rule_typestring—— 支持的规则类型全集rule_type是值校验的核心。目前支持的枚举定义在 AssertFieldRule.java 的AssertRuleType中与官方文档完全一致rule_type含义是否需要 rule_valueNOT_NULL值不能为 null否NULL值可以为 null否MIN数据的最小值数值下界含等号是MAX数据的最大值数值上界含等号是MIN_LENGTH字符串数据的最小长度是MAX_LENGTH字符串数据的最大长度是MIN_ROW最小行数是MAX_ROW最大行数是其中MIN_ROW/MAX_ROW属于行级规则应放在row_rules块中其余属于字段值规则放在field_rules的field_value中。注意源码中rule_value的类型是DoubleAssertRule.AssertRule中的private Double ruleValue因此rule_value需书写为数值。4.4 rule_valuenumeric与规则类型配套的值。当rule_type为MIN、MAX、MIN_LENGTH、MAX_LENGTH、MIN_ROW或MAX_ROW时必须为rule_value赋值。4.5 equals_toboolean | numeric | string | ConfigList | ConfigMap—— 精确值比较equals_to用于比较字段实际值是否等于配置的期望值支持所有 SeaTunnel 类型类型清单见 schema-feature 文档 的当前支持哪些类型一节。对于复杂类型需使用与上游数据一致的 HOCON 值形态。文档给出的示例某字段是包含三个字段的 row 类型声明为{a arraystring, b mapstring, decimal(30, 2), c{c_0 int, b string}}则期望值可以写成[[a, b], { k0 9999.99, k1 111.11 }, [123, abcd]]。两个重要注意点定义字段值的方式与 FakeSource 连接器 中定制数据内容的方式一致equals_to不能应用于null类型字段此时请改用规则类型NULL进行校验例如{rule_type NULL}。从实现看equals_to的比对并非简单的字符串相等AssertExecutor.compareValue会先把配置值通过JsonToRowConverters转换为 SeaTunnel 类型对象再按类型分发比对——ROW走逐字段递归比较、ARRAY走逐元素比较长度也必须一致、MAP比较 key 是否都存在且逐 value 递归比较、BYTES走Arrays.equals、其余标量类型走value.equals(confValue)。这意味着equals_to对arrayint、maptime, string、嵌套 row 等复杂值都能做结构化深度相等校验。4.6 catalog_table_ruleConfigMap—— Catalog 表元数据断言用于断言实际 Catalog 表元数据与用户定义的表元数据完全一致。它由四个子规则组成对应 AssertCatalogTableRule.java 中的四个字段primary_key_rule校验主键包含primary_key_name主键名与primary_key_columns主键列集合。源码中的比较逻辑为若期望主键名为空则跳过名比较若期望列集合非空则用CollectionUtils.isEqualCollection做无序集合相等比较constraint_key_rule校验约束键列表每项包含constraint_key_name、constraint_key_type如UNIQUE_KEY、constraint_key_columns每列含constraint_key_column_name与constraint_key_sort_type如ASC。源码同样使用isEqualCollection与实际的约束键集合比较column_rule按顺序逐列比较每列声明name、type、column_length、nullable、default_value、comment。源码isColumnEqual的比较维度包括列名、数据类型、列长度、scale、是否可空、默认值、注释、源类型任一维度不一致即抛错且要求列数完全一致table_identifier_rule源码中位于AssertConfig.TableIdentifierRule含catalog_name与table校验TableIdentifier与期望完全相等。4.7 table-namesConfigList—— 表名存在性断言用于断言输入数据中确实存在所列的表名。从 AssertSinkWriter.close() 的实现看Writer 在write阶段会把每行数据的tableId收集进静态集合TABLE_NAMESclose()时若配置了table-names会用new HashSet(期望表名).equals(TABLE_NAMES)做集合相等校验——注意这要求实际收集到的表名集合与期望集合完全相同不仅仅是包含关系。4.8 tables_configsConfigList—— 多表独立规则用于为多张输入表定义不同的断言规则每个元素必须包含table_path。table_path的取值必须与上游 Source 携带的表路径一致在 Zeta 引擎中上游表路径一般来自 Source 的tables_configs或schema.table声明。在多表作业中每条规则可以独立配置row_rules与field_rules。4.9 multi_table_sink_replicaint多表 Sink 的公共选项——每个表所需的 Sink 副本replica数量。仅在作业需要每张表多个 Sink 副本时配置。其余 Sink 公共参数参见 Sink 公共选项文档。五、规则匹配语义与执行时机5.1 三类规则的作用对象row_rules检查 Assert Sink 接收到的行数field_rules检查每一行中指定字段的值tables_configs用于多表作业table_path必须与上游 Source 携带的表路径匹配equals_to将实际字段值与配置的期望值比较复杂值array、map、row需使用与源数据相同的 HOCON 值形态。5.2 批式与流式下的执行时机重要Assert 同时支持BATCH与STREAMING两种作业模式但两条执行路径的时机完全不同这一点在原文档 sink/Assert.md 的 Streaming Validation 一节有明确说明且与源码实现完全吻合字段级规则逐行生效NOT_NULL、MIN_LENGTH、MAX_LENGTH、MIN、MAX、equals_to等在每行数据到达 Sink Writer 时立即检查AssertSinkWriter.write中调用ASSERT_EXECUTOR.fail(...)命中失败规则即抛出AssertConnectorException错误码为RULE_VALIDATION_FAILED消息形如row :row fail rule: rule行数规则仅在 Writer 关闭时执行一次MIN_ROW/MAX_ROW在 Sink Writerclose()时作业结束、savepoint 或失败时恰好执行一次针对的是该 Writer 实例自创建以来累计的行数——而非每个 checkpoint 窗口的行数也不会在 checkpoint 之间重置。累计计数通过静态ConcurrentHashMapString, LongAccumulatorLONG_ACCUMULATOR以表名为 key实现天然支持跨 checkpoint 累积。如果你的需求是按 checkpoint 窗口做行数校验这超出了当前文档与实现的能力范围需要修改源码才能支持。5.3 多表场景下的 close 语义AssertSinkWriter.close()对行数规则的校验逻辑有一个细节在多表作业中只有当assertRowRules.size() 1或规则的 key 与当前 Writer 的catalogTableName相等时才会执行该校验——也就是说每个 Writer 只对自己负责的那张表做行数断言。这一点由测试 AssertSinkWriterCloseTest.java 中的两个用例直接验证testCloseOnlyAssertsOwnTableWhenMultipleTablesWriter A 收到 1 行、表 B 从未收到行A 的MIN_ROW1校验必须通过不能被其他 Writer 的进度影响testCloseThrowsWhenOwnTableRuleNotMetA 自己只写了 1 行但MIN_ROW2close 时必须抛AssertConnectorException。六、实战示例6.1 简单示例行数 字段规则 Catalog 元数据下面的作业验证管道输出行数在 510 之间且name字段非空、字符串长度在 510age字段非空、精确等于 23、值域在 327672147483647同时断言 Catalog 表的主键、唯一约束与列元数据。Assert { rules { row_rules [ { rule_type MAX_ROW rule_value 10 }, { rule_type MIN_ROW rule_value 5 } ], field_rules [{ field_name name field_type string field_value [ { rule_type NOT_NULL }, { rule_type MIN_LENGTH rule_value 5 }, { rule_type MAX_LENGTH rule_value 10 } ] }, { field_name age field_type int field_value [ { rule_type NOT_NULL equals_to 23 }, { rule_type MIN rule_value 32767 }, { rule_type MAX rule_value 2147483647 } ] } ] catalog_table_rule { primary_key_rule { primary_key_name primary key primary_key_columns [id] } constraint_key_rule [ { constraint_key_name unique_name constraint_key_type UNIQUE_KEY constraint_key_columns [ { constraint_key_column_name id constraint_key_sort_type ASC } ] } ] column_rule [ { name id type bigint }, { name name type string }, { name age type int } ] } } }注意catalog_table_rule是对Catalog 元数据的断言示例中的column_rule必须与实际管道中表的列定义逐维一致名称、类型、长度、可空性、默认值、注释等否则作业会以CATALOG_TABLE_FAILED错误码失败。6.2 复杂示例equals_to 全类型精确比对下面的完整作业先用 FakeSource 产出一条包含 null、string、boolean、整数族、浮点族、decimal、date/timestamp/time、bytes、array、map含嵌套 map与嵌套 row 的记录再用 Assert 的equals_to逐字段做精确比对。这是equals_to最典型的全类型验证场景。FakeSource 的用法参见 FakeSource 文档。source { FakeSource { row.num 1 schema { fields { c_null null c_string string c_boolean boolean c_tinyint tinyint c_smallint smallint c_int int c_bigint bigint c_float float c_double double c_decimal decimal(30, 8) c_date date c_timestamp timestamp c_time time c_bytes bytes c_array arrayint c_map maptime, string c_map_nest mapstring, {c_int int, c_string string} c_row { c_null null c_string string c_boolean boolean c_tinyint tinyint c_smallint smallint c_int int c_bigint bigint c_float float c_double double c_decimal decimal(30, 8) c_date date c_timestamp timestamp c_time time c_bytes bytes c_array arrayint c_map mapstring, string } } } rows [ { kind INSERT fields [ null, AAA, false, 1, 1, 333, 323232, 3.1, 9.33333, 99999.99999999, 2012-12-21, 2012-12-21T12:34:56, 12:34:56, bWlJWmo, [0, 1, 2], { 12:01:26 v0 }, { k1 [123, BBB-BB]}, [ null, AAA, false, 1, 1, 333, 323232, 3.1, 9.33333, 99999.99999999, 2012-12-21, 2012-12-21T12:34:56, 12:34:56, bWlJWmo, [0, 1, 2], { k0 v0 } ] ] } ] plugin_output fake } } sink{ Assert { plugin_input fake rules { row_rules [ { rule_type MAX_ROW rule_value 1 }, { rule_type MIN_ROW rule_value 1 } ], field_rules [ { field_name c_null field_type null field_value [ { rule_type NULL } ] }, { field_name c_string field_type string field_value [ { rule_type NOT_NULL equals_to AAA } ] }, { field_name c_boolean field_type boolean field_value [ { rule_type NOT_NULL equals_to false } ] }, { field_name c_tinyint field_type tinyint field_value [ { rule_type NOT_NULL equals_to 1 } ] }, { field_name c_smallint field_type smallint field_value [ { rule_type NOT_NULL equals_to 1 } ] }, { field_name c_int field_type int field_value [ { rule_type NOT_NULL equals_to 333 } ] }, { field_name c_bigint field_type bigint field_value [ { rule_type NOT_NULL equals_to 323232 } ] }, { field_name c_float field_type float field_value [ { rule_type NOT_NULL equals_to 3.1 } ] }, { field_name c_double field_type double field_value [ { rule_type NOT_NULL equals_to 9.33333 } ] }, { field_name c_decimal field_type decimal(30, 8) field_value [ { rule_type NOT_NULL equals_to 99999.99999999 } ] }, { field_name c_date field_type date field_value [ { rule_type NOT_NULL equals_to 2012-12-21 } ] }, { field_name c_timestamp field_type timestamp field_value [ { rule_type NOT_NULL equals_to 2012-12-21T12:34:56 } ] }, { field_name c_time field_type time field_value [ { rule_type NOT_NULL equals_to 12:34:56 } ] }, { field_name c_bytes field_type bytes field_value [ { rule_type NOT_NULL equals_to bWlJWmo } ] }, { field_name c_array field_type arrayint field_value [ { rule_type NOT_NULL equals_to [0, 1, 2] } ] }, { field_name c_map field_type maptime, string field_value [ { rule_type NOT_NULL equals_to { 12:01:26 v0 } } ] }, { field_name c_map_nest field_type mapstring, {c_int int, c_string string} field_value [ { rule_type NOT_NULL equals_to { k1 [123, BBB-BB] } } ] }, { field_name c_row field_type { c_null null c_string string c_boolean boolean c_tinyint tinyint c_smallint smallint c_int int c_bigint bigint c_float float c_double double c_decimal decimal(30, 8) c_date date c_timestamp timestamp c_time time c_bytes bytes c_array arrayint c_map mapstring, string } field_value [ { rule_type NOT_NULL equals_to [ null, AAA, false, 1, 1, 333, 323232, 3.1, 9.33333, 99999.99999999, 2012-12-21, 2012-12-21T12:34:56, 12:34:56, bWlJWmo, [0, 1, 2], { k0 v0 } ] } ] } ] } } }从实现角度补充两个可验证的细节见 AssertExecutor.javaDecimal 校验同时检查精度与小数位checkDecimalType会校验实际值的scale()必须等于声明的 scale且precision()必须小于等于声明的 precision否则类型检查不通过NULL 类型字段的强约束checkType中若字段类型为SqlType.NULL而值非空直接返回 false反之若字段声明了具体类型而值为 null类型检查本身放行是否允许 null 交由rule_type NULL/NOT_NULL决定——这就是为什么c_null字段要用rule_type NULL校验而不能用equals_to。6.3 多表断言每张表独立的行数与字段规则下面的作业在同一个 Job 中处理两张表test.table116 行c_int/c_bigint与test.table217 行c_string/c_tinyintAssert 通过tables_configs对每张表分别断言行数与字段非空。env { parallelism 1 job.mode BATCH } source { FakeSource { tables_configs [ { row.num 16 schema { table test.table1 fields { c_int int c_bigint bigint } } }, { row.num 17 schema { table test.table2 fields { c_string string c_tinyint tinyint } } } ] } } transform { } sink { Assert { rules { tables_configs [ { table_path test.table1 row_rules [ { rule_type MAX_ROW rule_value 16 }, { rule_type MIN_ROW rule_value 16 } ], field_rules [{ field_name c_int field_type int field_value [ { rule_type NOT_NULL } ] }, { field_name c_bigint field_type bigint field_value [ { rule_type NOT_NULL } ] }] }, { table_path test.table2 row_rules [ { rule_type MAX_ROW rule_value 17 }, { rule_type MIN_ROW rule_value 17 } ], field_rules [{ field_name c_string field_type string field_value [ { rule_type NOT_NULL } ] }, { field_name c_tinyint field_type tinyint field_value [ { rule_type NOT_NULL } ] }] } ] } } }6.4 流式校验示例累积行数窗口 字段值域下面的流式作业以 60 秒为 checkpoint 间隔运行FakeSource 产出 1000 行数据。Assert 在 Writer 关闭时对累积行数做50 ≤ total rows ≤ 5000的校验注意不是按 checkpoint 窗口同时对每行的age字段执行非空与0 ≤ age ≤ 150的逐行校验。env { parallelism 1 job.mode STREAMING checkpoint.interval 60000 } source { FakeSource { row.num 1000 schema { fields { name string age int } } plugin_output stream_data } } sink { Assert { plugin_input stream_data rules { row_rules [ { rule_type MIN_ROW rule_value 50 }, { rule_type MAX_ROW rule_value 5000 } ], field_rules [{ field_name age field_type int field_value [ { rule_type NOT_NULL }, { rule_type MIN rule_value 0 }, { rule_type MAX rule_value 150 } ] }] } } }七、源码级原理剖析7.1 单行校验链路AssertSinkWriter → AssertExecutorAssertSinkWriter.write(SeaTunnelRow)AssertSinkWriter.java是整条校验链路的入口核心逻辑可以归纳为三步记录表名与累计行数把当前行的tableId加入静态集合TABLE_NAMES用LONG_ACCUMULATOR.computeIfAbsent(tableName, k - new LongAccumulator(Long::sum, 0)).accumulate(1)对该表的行数累加 1定位字段规则单表场景直接取唯一一份field_rules多表场景按行的tableId或catalogTableName从MapString, ListAssertFieldRule中取出对应规则执行字段规则调用静态单例ASSERT_EXECUTOR.fail(element, seaTunnelRowType, rules)一旦OptionalAssertFieldRule非空立即抛出AssertConnectorExceptionRULE_VALIDATION_FAILED。7.2 规则执行器AssertExecutor 的判定顺序AssertExecutorAssertExecutor.java对每个字段的判定顺序是先类型后值。checkType按 SQL 类型分发——ROW递归检查每个子字段、ARRAY检查每个元素、MAP同时检查 key 与 value、DECIMAL检查精度与小数位、向量类型要求ByteBuffer、其余类型要求value.getClass().equals(fieldType.getTypeClass())checkValue逐条执行field_value中的规则。rule_type不为空时先按checkAssertRule的 switch 判定NULL/NOT_NULL判空、MIN/MAX用doubleValue()比较、MIN_LENGTH/MAX_LENGTH用字符串长度比较随后若配置了equals_to且值非空再执行compareValue做结构化深度比较。7.3 行数规则close 时的一次性断言AssertSinkWriter.close()中row_rulesMIN_ROW/MAX_ROW只在该 Writer 负责的表上执行一次从LONG_ACCUMULATOR取出该表累计行数若count MAX_ROW或count MIN_ROW抛出携带实际行数的AssertConnectorException消息形如row num :count fail rule: rule。table-names的集合相等校验也在此处完成。7.4 Catalog 元数据断言AssertCatalogTableRuleAssertCatalogTableRule.checkRule(CatalogTable)AssertCatalogTableRule.java依次校验主键、约束键、列与表标识符任何不一致都以CATALOG_TABLE_FAILED错误码失败。列比较isColumnEqual覆盖了名称、数据类型、列长度、scale、可空性、默认值、注释、源类型共 8 个维度。八、版本演进时间线ChangelogAssert 连接器自 2.2.0-beta 引入以来持续演进各版本的关键能力如下依据 connector-assert changelog 整理2.2.0-beta将 Assert Sink 加入 API 草案add assert sink to Api draft随后以统一工厂Source/Sink Factory形式完善连接器框架2.3.0 / 2.3.0-beta为 Assert 增加 Sink 工厂并统一异常体系Unified exception保证factoryIdentifier与插件名一致2.3.1配合 SimpleSQL 等 Transform 演进完善多模块构建2.3.4能力密集版本支持全数据类型的字段类型断言与字段值相等断言field type assert field value equality assert for full data types支持检查 Decimal 类型的 precision 与 scale新增table-names以支持 FakeSource/Assert 的多表产出与断言schema 支持配置 column/primaryKey/constraintKey2.3.5修复 DateTime 工具相关问题2.3.6支持在 Sink 选项中使用上游表占位符并自动替换2.3.7增加多表 Sink 选项检查2.3.8支持多表校验Assert support multi-table check2.3.9支持带时区偏移的时间戳timestamp with timezone offset优化 Assert Sink 校验方法统一tables_configs与table_list2.3.10新增 Assert options重构 connector common options2.3.12将元数据 schema 引入 Catalog 表add metadata schema into catalog table。九、常见误区与使用建议行数规则不是逐行触发的MIN_ROW/MAX_ROW只在 Writer 关闭时执行一次且统计的是该 Writer 自创建以来的累积行数。若在流式作业中期望每个 checkpoint 窗口校验行数当前版本无法直接满足需要源码级改动equals_to对 null 字段无效null 类型字段请用rule_type NULL/NOT_NULL控制table-names是集合相等而非包含期望表名集合必须与实际接收到的表名集合完全一致否则 close 时失败catalog_table_rule是全维度逐列比对列数、列名、类型、长度、scale、可空性、默认值、注释、源类型任一不一致都会使作业失败配置前请确认上游 Catalog 元数据至少配置一条有意义的规则只有rules是必填但空规则集意味着无事可校验不会起到任何质量保障作用table_path必须与上游一致多表场景下tables_configs中的table_path必须与 Source 携带的表路径精确匹配。十、相关资源Assert 连接器主文档docs/en/connectors/sink/Assert.md变更日志docs/en/connectors/changelog/connector-assert.md源码目录seatunnel-connectors-v2/connector-assert含sink/、rule/、excecutor/、exception/子包单元测试AssertSinkWriterCloseTest.java多表 close 语义、AssertExecutorTest.java配套的 FakeSource 文档docs/en/connectors/source/FakeSource.mdequals_to值形态定义方式与其一致Sink 公共选项docs/en/connectors/common-options/sink-common-options.mdSchema 类型声明指南docs/en/introduction/concepts/schema-feature.md【免费下载链接】seatunnelSeaTunnel is a multimodal, high-performance, distributed, massive data integration tool.项目地址: https://gitcode.com/GitHub_Trending/se/seatunnel创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考