Flink 1.7流环境下无需重启作业动态修改SQL中botcode值的方法

因为Flink 1.7版本还没有支持动态SQL更新的原生能力（比如后续版本的SQL Client动态提交或者动态表功能），所以我们可以通过可动态更新的UDF+配置中心的方式实现无需重启作业修改botcode匹配值的需求，具体方案如下：

核心思路 #

把SQL中硬编码的botcode='r1'这类判断逻辑抽离到一个自定义UDF中，让UDF从外部配置中心（比如ZooKeeper、Nacos，甚至简单的HTTP接口）实时获取目标匹配值。这样不用修改SQL语句，只需要更新配置中心的值，UDF就会自动应用新的匹配规则。

具体实现步骤 #

1. 编写支持动态配置的Scalar UDF #

这个UDF会维护当前的目标botcode值，并且定期（或通过监听）从配置中心拉取最新值，确保线程安全的情况下提供匹配判断：

import org.apache.flink.table.functions.ScalarFunction;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

public class DynamicBotCodeMatcher extends ScalarFunction {
    // 使用volatile保证多线程下的可见性
    private volatile String targetBotCode = "r1";
    private ScheduledExecutorService configRefreshScheduler;

    @Override
    public void open(FunctionContext context) throws Exception {
        super.open(context);
        // 初始化定时任务，每30秒拉取一次配置（可根据需求调整间隔）
        configRefreshScheduler = Executors.newSingleThreadScheduledExecutor();
        configRefreshScheduler.scheduleAtFixedRate(
            this::refreshTargetBotCode,
            0, // 首次执行延迟
            30, // 刷新间隔
            TimeUnit.SECONDS
        );
    }

    // 从配置中心拉取最新目标值，这里需要替换为你的配置中心实现
    private void refreshTargetBotCode() {
        // 示例：从ZooKeeper/Nacos/HTTP接口获取值
        String newTargetCode = ConfigCenter.fetchLatestBotCode("icf");
        if (newTargetCode != null && !newTargetCode.isEmpty()) {
            this.targetBotCode = newTargetCode;
        }
    }

    // UDF核心逻辑：判断传入的botcode是否匹配目标值
    public Integer eval(String inputBotCode) {
        return targetBotCode.equals(inputBotCode) ? 1 : 0;
    }

    @Override
    public void close() throws Exception {
        super.close();
        // 关闭定时任务
        configRefreshScheduler.shutdown();
    }
}

// 模拟配置中心工具类，实际项目中替换为真实实现
class ConfigCenter {
    public static String fetchLatestBotCode(String key) {
        // 这里写你的配置读取逻辑，比如从ZooKeeper节点读取、调用HTTP接口等
        // 示例：假设用户通过配置中心设置了icf对应的botcode为'r10'
        return "r10";
    }
}

2. 注册UDF并修改SQL语句 #

在你的Flink作业代码中，先把UDF注册到TableEnvironment，然后修改原SQL，用UDF调用替代硬编码的判断：

// 假设你已经初始化了TableEnvironment
TableEnvironment tableEnv = ...;

// 注册自定义UDF
tableEnv.registerFunction("match_bot_code", new DynamicBotCodeMatcher());

// 修改后的SQL语句
String ipdetailsSql = "select sid, _zpsbd6 as ip_address, ssresp, reason, " + 
    "SUM(match_bot_code(botcode)) as icf_count, " + 
    "SUM(CASE WHEN botcode='r2' THEN 1 ELSE 0 END ) as dc_count, " + 
    "SUM(CASE WHEN botcode='r5' THEN 1 ELSE 0 END ) as badua_count, " + 
    "COUNT(*) as hits, TUMBLE_START(ts, INTERVAL '1' MINUTE) AS fseen " + 
    "from sourceTopic " + 
    "GROUP BY TUMBLE(ts, INTERVAL '1' MINUTE), sid, _zpsbd6, ssresp, reason";

3. 优化：实时配置监听（可选） #

如果需要更实时的配置更新，可以把定时轮询改成配置中心的事件监听（比如ZooKeeper的Watcher、Nacos的配置监听），这样配置一更新，UDF就能立刻感知到。以ZooKeeper为例，修改UDF的open方法：

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.WatchedEvent;

@Override
public void open(FunctionContext context) throws Exception {
    super.open(context);
    // 初始化ZooKeeper客户端
    CuratorFramework zkClient = CuratorFrameworkFactory.newClient(
        "your-zk-host:2181",
        new ExponentialBackoffRetry(1000, 3)
    );
    zkClient.start();

    // 监听配置节点变化
    String configPath = "/flink/config/botcode/icf";
    zkClient.getData().usingWatcher(new Watcher() {
        @Override
        public void process(WatchedEvent event) {
            if (event.getType() == Event.EventType.NodeDataChanged) {
                try {
                    // 获取最新配置值
                    byte[] data = zkClient.getData().forPath(configPath);
                    targetBotCode = new String(data, StandardCharsets.UTF_8);
                    // 重新注册Watcher（ZooKeeper的Watcher是一次性的）
                    zkClient.getData().usingWatcher(this).forPath(configPath);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }).forPath(configPath);
}

扩展：支持多botcode动态修改 #

如果后续需要动态修改r2、r5这类其他botcode的匹配值，可以修改UDF，让它接受一个类型参数，从配置中心按key拉取对应的值：

public Integer eval(String inputBotCode, String matcherType) {
    String targetCode = ConfigCenter.fetchLatestBotCode(matcherType);
    return targetCode != null && targetCode.equals(inputBotCode) ? 1 : 0;
}

对应的SQL修改为：

"SUM(match_bot_code(botcode, 'icf')) as icf_count, " + 
"SUM(match_bot_code(botcode, 'dc')) as dc_count, " + 
"SUM(match_bot_code(botcode, 'badua')) as badua_count, " + 

这样你只需要在配置中心分别维护icf、dc、badua对应的botcode值即可。

注意事项 #

• 确保UDF中的配置变量用volatile修饰，保证多并行实例之间的可见性
• 配置中心的读取逻辑要保证异常容错，避免因为配置读取失败导致作业异常
• Flink 1.7的UDF需要继承ScalarFunction，确保符合版本要求

内容的提问来源于stack exchange，提问作者Ravi Shanker Reddy