请教下在聚合场景下，如果上游数据出现更新或者删除，如何保证flink sql聚合结果的准确性？

在Flink SQL的聚合场景中，如果上游数据出现更新或删除操作（如UPDATE_BEFORE、DELETE等），为了保证聚合结果的准确性，需要结合Flink SQL的机制和参数配置进行优化。以下是具体的解决方案和建议：

1. 启用MiniBatch优化

MiniBatch是一种通过缓存一定量的数据后再触发处理的机制，可以减少对State的频繁访问，从而提升吞吐并降低延迟。对于包含更新或删除操作的场景，MiniBatch能够有效减少状态更新的频率，确保聚合结果的准确性。

• 开启方式
  在作业运行参数中添加以下配置：

  table.exec.mini-batch.enabled: true
  table.exec.mini-batch.allow-latency: 5s
  

  • table.exec.mini-batch.enabled：是否开启MiniBatch。
  • table.exec.mini-batch.allow-latency：批量输出数据的时间间隔。

• 注意事项

  • MiniBatch通过增加延迟换取高吞吐，因此不适用于超低延迟要求的场景。
  • 如果聚合结果变化周期小于State TTL设置时间，则无需额外配置table.exec.agg.mini-batch.output-identical-enabled参数。

2. 启用LocalGlobal优化

LocalGlobal优化将聚合分为两个阶段：局部聚合（LocalAgg）和全局聚合（GlobalAgg）。这种分阶段的处理方式可以有效缓解数据热点问题，并确保在更新或删除操作下聚合结果的准确性。

• 适用场景

  • 普通聚合操作（如SUM、COUNT、MAX、MIN、AVG）。
  • 数据热点问题较为明显的场景。

• 使用限制

  • LocalGlobal优化需要在MiniBatch开启的前提下才能生效。
  • 需要使用AggregateFunction实现Merge操作。

• 判断是否生效
  观察最终生成的拓扑图节点名称中是否包含GlobalGroupAggregate或LocalGroupAggregate。

3. 处理Changelog事件乱序问题

在Flink SQL中，上游数据的更新或删除操作可能会导致Changelog事件乱序，从而影响聚合结果的准确性。为了解决这一问题，可以通过配置SinkUpsertMaterializer来确保数据的物理化处理。

• 参数设置
  使用table.exec.sink.upsert-materialize参数控制是否启用SinkUpsertMaterializer：

  • auto（默认值）：Flink会根据正确性推断是否需要添加SinkUpsertMaterializer。
  • none：不使用SinkUpsertMaterializer。
  • force：强制使用SinkUpsertMaterializer。

• 避免使用SinkUpsertMaterializer的建议

  • 确保去重、分组聚合等操作的分区键与结果表的主键一致。
  • 如果下游算子与上游的去重、分组聚合相连，且在VVR 6.0以下版本中未出现数据准确性问题，可以将table.exec.sink.upsert-materialize设置为none。

• 注意事项

  • 避免在写入结果表时添加由非确定性函数（如CURRENT_TIMESTAMP、NOW）生成的列，否则可能导致状态异常膨胀。
  • 如果已出现大状态问题，可通过增加作业并发度来缓解性能压力。

4. 合理设置State TTL

State TTL（Time-To-Live）用于控制状态的有效期，避免状态无限增长。在聚合场景中，合理设置State TTL可以防止因长时间未收到更新而导致的状态过期问题。

• 参数说明

  • table.exec.agg.mini-batch.output-identical-enabled：控制在开启State TTL且聚合结果未发生变化的情况下，是否仍然向下游发送重复数据。
  • 默认值为false：聚合结果未变化时不向下游发送数据。
  • 设置为true：即使聚合结果未变化，仍向下游发送更新数据。

• 注意事项

  • 该参数仅在VVR 8.0.8及以上版本生效。
  • 将参数从false修改为true可能会增加下游算子的压力。

5. 忽略Delete操作

在某些场景下，可以通过忽略DELETE操作来避免因删除操作导致的数据不一致问题。

• 参数设置
  使用sink.ignore-delete参数控制是否忽略DELETE操作：

  • 默认值为false：不忽略DELETE操作。
  • 设置为true：忽略DELETE操作。

• 适用场景

  • 多个输出节点根据主键同时更新同一张结果表的不同字段的场景。
  • 删除操作可能导致部分字段被置为null或默认值的场景。

6. 调整写入策略

为了进一步优化写入性能并确保聚合结果的准确性，可以调整写入策略相关参数。

• 关键参数

  • sink.delete-strategy：定义撤回消息的处理策略。
  • IGNORE_DELETE：忽略Update Before和Delete消息。
  • NON_PK_FIELD_TO_NULL：将Delete消息执行为将非主键字段更新为null。
  • DELETE_ROW_ON_PK：根据主键删除整行。
  • CHANGELOG_STANDARD：按照Flink SQL Changelog的工作原理运行，确保数据准确性。
  • jdbcWriteBatchSize、jdbcWriteBatchByteSize、jdbcWriteFlushInterval：控制数据攒批写入的条件。

• 注意事项

  • 攒批写入虽然能提升写入效率，但可能带来数据延迟问题和内存溢出风险，需根据实际业务场景权衡。

总结

通过上述方法，可以在Flink SQL聚合场景中有效应对上游数据的更新或删除操作，确保聚合结果的准确性。具体包括启用MiniBatch和LocalGlobal优化、处理Changelog事件乱序问题、合理设置State TTL、忽略Delete操作以及调整写入策略。根据实际业务需求选择合适的配置方案，可显著提升作业的稳定性和性能。