本文共 1135 字，大约阅读时间需要 3 分钟。

Flink SQL 运维与优化指南

一、需求与问题

需求一：数据汇总优化

在 Flink SQL 中，每1分钟汇总一次当天数据。频繁的写入操作导致数据库记录过多，建议开启 minibatch 模式，以提高处理效率。具体设置如下：

• settable.exec.mini-batch.enabled= true;
• settable.exec.mini-batch.allow-latency= 10s;
• settable.exec.mini-batch.size= 100000;

需求二：数据持久化

Flink SQL 的 GROUP BY 操作生成的数据是 CDC 类型，无法直接删除。建议启用动态表选项，避免数据丢失：

• settable.dynamic-table-options.enabled= true;
• 使用 /*+OPTIONS('maxwell-json-ext.ignore-delete' = 'true', 'maxwell-json-ext.update-to-insert' = 'true')*/ 忽略删除操作。

需求三：自动生成测试数据

使用 DataGen 连接器在 Flink 1.11 中生成随机测试数据。详细配置方法请参考 官方文档。

需求四：字符串分割处理

实现自定义表函数，将字符串分割为多列：

• 创建 SplitFunction 类，继承 TableFunction。
• 使用 str.split(" ") 分割字符串，输出每个分割结果及其长度。
• 注册表函数并在 SQL 中调用。

需求五：字段空值处理

在 Flink SQL 中，使用 cast(null as dataType) 来表示字段为空。

需求六：数据恢复

平台无法正常启动时，设置 Kafka 消费起始时间：

• set 'scan.startup.mode' = 'timestamp';
• set 'scan.startup.timestamp-millis' = '1617551100000';

二、问题与解决方案

问题一：类型映射错误

MySQL tinyint 类型在 Flink 上可能映射失败。修改数据库连接参数：

• URL 格式：jdbc:mysql://...?tinyInt1isBit=false

问题二：Flink 背压问题

详细解决方案请参考 技术博客。

问题三：状态膨胀

解决方法请参考 腾讯云开发文章。

问题四：最终结果丢失

通过双写至 HBase 对比数据，发现问题通常出在 MySQL 主键定义不一致。确保 Flink 平台和数据库主键一致。

三、总结

以上内容涵盖了 Flink SQL 的常见需求与问题解决方案，希望对您的项目有所帮助。