以API为数据源的Google Dataflow生产环境超时问题求助

核心问题是Google Dataflow Flex模板的构建阶段存在固定超时（约12-13分钟），而当前实现是在管道构建阶段同步拉取全量192K条API数据到内存List，这个过程耗时超过了超时阈值，导致Dataflow直接终止任务。Cloud Composer的execution_timeout控制的是DAG整体执行时长，无法覆盖Dataflow模板构建阶段的超时限制。

1. 短期快速修复：优化同步拉取效率 #

如果不想大幅改动代码，可以先通过优化拉取逻辑压缩总耗时：

• 并行拉取分页数据：用线程池同时拉取多个分页（比如10-20个并行请求，需注意API并发限制），替代串行循环拉取，减少总等待时间。
• 调大单次拉取条数：如果API支持，尝试将每页拉取量从1000条调至更大值，减少总请求次数。
• 复用HTTP连接：启用HttpClient连接池，避免每次请求重新建立连接，降低请求 overhead。

2. 根本解决：将数据拉取移至管道执行阶段 #

推荐采用这种方案，彻底避开构建阶段的超时限制：核心思路是不在管道初始化时拉取数据，而是将分页标记作为管道输入，在Worker节点上执行数据拉取。

实现示例 #

方式1：手动生成分页标记

// 构建阶段仅计算并生成页码列表（耗时极短）
int totalPages = (int) Math.ceil(192000.0 / 1000);
List<Integer> pageNumbers = IntStream.rangeClosed(1, totalPages).boxed().collect(Collectors.toList());

// 管道执行阶段由Worker拉取对应页数据
PCollection<String> records = pipeline.apply(Create.of(pageNumbers))
    .apply(ParDo.of(new DoFn<Integer, String>() {
        @ProcessElement
        public void processElement(ProcessContext c) {
            int currentPage = c.element();
            // 根据页码拉取单页API数据
            List<String> pageRecords = fetchApiPage(currentPage, 1000);
            // 展开单页数据为单条记录输出
            pageRecords.forEach(c::output);
        }
    }));

// 后续写入GCS的逻辑保持不变
records.apply(TextIO.write().to("gs://your-bucket/path"));

方式2：动态生成分页标记

如果总记录数不确定，可先快速拉取一次API获取总数，再用GenerateSequence生成页码流：

// 构建阶段仅拉取总记录数（单次请求耗时短）
int totalRecords = fetchApiTotalCount();
int totalPages = (int) Math.ceil(totalRecords / 1000.0);

PCollection<String> records = pipeline.apply(GenerateSequence.from(1).to(totalPages))
    .apply(ParDo.of(new FetchPageDoFn()));

这种方案的优势：

• 管道构建阶段几乎无耗时，完全避开超时问题
• 天然支持并行拉取（Dataflow会自动分配不同页码到多个Worker），拉取效率更高
• 无需实现复杂的自定义Source的split()方法

3. 自定义Source（适合长期复用场景） #

如果坚持要实现自定义Source，可简化split()逻辑：

• 先拉取总记录数计算总页数
• split()方法直接将每个页码作为一个分片（Shard），每个分片对应一页数据
• Reader仅需根据分片内的页码拉取对应页数据即可，无需处理复杂的范围拆分

优先选择方案2，既能彻底解决超时问题，又能提升数据拉取的并行效率，实现成本远低于自定义Source。

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