Apache Flink：如何实现持续获取数据的SourceFunction？

嘿，我来帮你搞定这个问题！你的核心困扰是自定义的MySource只拉取了一次URL数据就停了，导致下游的10分钟窗口只能处理单次数据。要让Source持续跑起来，只需要给它加个循环拉取的逻辑，同时处理任务取消的信号就行。

问题根源 #

你的MySource应该是在run(SourceContext<T> ctx)方法里只写了一次数据获取代码，没有做循环拉取的处理，所以Source执行完一次就终止了，自然没法给窗口持续喂数据。

解决方案：给Source加循环拉取逻辑 #

要让Source持续运行，你需要在run方法里完成这几件事：

• 用while循环包裹数据拉取逻辑，让它重复执行
• 设置一个标志位控制循环启停，配合cancel()方法处理任务取消
• 每次拉取后发送数据到下游，再设置间隔时间避免频繁请求

完整的MySource实现代码 #

import org.apache.flink.streaming.api.functions.source.SourceFunction;

public class MySource implements SourceFunction<String> {
    // 用volatile保证多线程下的状态可见性，控制Source是否运行
    private volatile boolean isRunning = true;
    // 拉取间隔，比如每隔1分钟拉一次（可根据业务需求调整）
    private final long pullInterval = 60 * 1000;

    @Override
    public void run(SourceContext<String> ctx) throws Exception {
        // 循环拉取，直到任务被取消
        while (isRunning) {
            // 替换成你实际的URL数据获取逻辑，比如用HttpClient发请求
            String data = fetchDataFromTargetUrl();
            
            // 将获取到的数据发送给下游算子
            ctx.collect(data);
            
            // 等待指定间隔后再拉取下一批
            Thread.sleep(pullInterval);
        }
    }

    @Override
    public void cancel() {
        // 任务取消时，修改标志位终止循环
        isRunning = false;
    }

    // 模拟从URL获取数据的方法，替换成你的真实实现
    private String fetchDataFromTargetUrl() {
        // 这里写你的HTTP请求代码，示例如下：
        // try (CloseableHttpClient client = HttpClients.createDefault()) {
        //     HttpGet request = new HttpGet("https://your-target-url.com");
        //     try (CloseableHttpResponse response = client.execute(request)) {
        //         return EntityUtils.toString(response.getEntity());
        //     }
        // } catch (Exception e) {
        //     throw new RuntimeException("拉取数据失败", e);
        // }
        return "从URL获取的字符串数据";
    }
}

主程序代码调整（保留你的窗口逻辑） #

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

public class StreamingJob {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<String> dataStream = env.addSource(new MySource())
                .keyBy(value -> {
                    // 替换成你的自定义KeyBy逻辑，比如根据数据中的某个字段分组
                    return "你的自定义Key";
                })
                .window(TumblingProcessingTimeWindows.of(Time.minutes(10)))
                // 这里添加你的窗口处理逻辑，比如聚合、计算等
                .reduce((v1, v2) -> String.format("%s;%s", v1, v2));

        dataStream.print();

        env.execute("持续拉取URL数据的窗口任务");
    }
}

额外注意事项 #

• 异常处理：如果URL拉取可能失败，建议在循环内加try-catch，避免单次拉取异常导致整个Source崩溃：

  while (isRunning) {
      try {
          String data = fetchDataFromTargetUrl();
          ctx.collect(data);
      } catch (Exception e) {
          System.err.println("拉取URL数据失败，跳过本次拉取：" + e.getMessage());
      }
      Thread.sleep(pullInterval);
  }
  

• 拉取间隔：根据你的业务需求调整pullInterval，不要太频繁给目标URL造成压力，也不要间隔太长影响数据时效性。
• 并行度：如果需要更高的拉取吞吐量，可以考虑将Source改为并行Source（实现ParallelSourceFunction），但要注意目标URL的并发请求限制。

内容的提问来源于stack exchange，提问作者Kspace