Apache Beam/Dataflow读取日期分区GCS动态路径的类型不匹配问题

• API调用产生的失败记录按天分区存储在GCS存储桶中，路径格式示例如下：

gs://path/to/file/2022/07/01
gs://path/to/file/2022/07/02

其余日期路径按相同规则生成。

• 需求：基于Apache Beam和Dataflow调度批处理作业，在次日重试对应日期分区下的失败记录。

GCS路径末尾的日期在初始模板上传到GCP时就被固定，后续无论何时运行作业都不会自动更新，该场景与Dataflow模板动态传入运行时日期参数的已知问题一致。
测试发现管道在本地运行时逻辑正常，但部署到Dataflow运行时，只有expand函数内的DoFn会被实际执行。最初尝试使用ValueProvider实现动态日期传入，未找到可行方案。

原有实现逻辑如下：

1. 在fileMatch阶段获取初始GCS路径，在路径末尾拼接当前时间往前24小时的日期
2. 调用FileIO.readMatches()将每个match()的结果转换为ReadableFile
3. 调用自定义MatchGCSFiles逻辑，通过ValueProvider获取当前日期并拼接至GCS路径，覆盖原有路径（因最初认为DoFn无法接收空输入，选择了该方案）
4. 再次调用FileIO.readMatches()将新的match()结果转换为新的ReadableFile，之后调用API执行重试逻辑

对应代码片段如下：

String dateFormat = "yyyy/MM/dd";
ValueProvider<String> date =
    new ValueProvider<String>() {
      @Override
      public String get() {
        String currentDate = Instant.now().minus(86400000).toDateTime(DateTimeZone.UTC).toString(dateFormat);
        return currentDate;
      }

      @Override
      public boolean isAccessible() {
        return true;
      }
    };
ValueProvider<String> gcsPathWithDate =
    new ValueProvider<String>() {
      @Override
      public String get() {
        return String.format("%s/%s/*/*.json", gcsPathPrefix, date.get());
      }

      @Override
      public boolean isAccessible() {
        return true;
      }
    };
fileMatch = FileIO.match().filepattern(gcsPathWithDate.get());
}

PCollectionTuple mixedPColl =
    input
        .getPipeline()
        .apply("File match", fileMatch)
        .apply("applying read matches", FileIO.readMatches())
        .apply("matching files", ParDo.of(new MatchGCSFiles()))
        .apply("applying read matches", FileIO.readMatches()) // 报错位置
        .apply("Read failed events from GCS", ParDo.of(new ReadFromGCS()))
        .apply(/* 调用API逻辑 */)...

当前报错出现在第二次调用FileIO.readMatches()的位置，返回类型不匹配，报错信息为：reason: no instance(s) of type variable(s) exist so that PCollection conforms to PCollection，尝试多种规避方案均未生效。

现有实现存在两个核心错误：

1. 自定义ValueProvider写法无效：模板生成阶段会直接调用gcsPathWithDate.get()拿到固定路径，后续作业运行时不会重新执行get()方法内的日期计算逻辑，这是路径被固定的根本原因。
2. IO组件类型不匹配：FileIO.readMatches()要求输入必须是PCollection<MatchResult.Metadata>类型，自定义MatchGCSFiles DoFn输出的不是该类型，自然会报类型不匹配错误，两次调用match和readMatches属于冗余逻辑，完全不需要。

核心原则：不要在管道构造阶段（即apply方法调用的外层代码位置）编写任何动态计算逻辑，这类代码只会在生成模板、本地提交作业时执行一次，部署到Dataflow运行时不会重新执行。所有动态逻辑要么写在DoFn内部，要么通过ValueProvider传入支持运行时参数的IO组件，不要提前调用ValueProvider.get()。

方案1：运行时传入日期参数（生产环境推荐） #

不在代码中硬编码日期计算逻辑，将日期作为模板的运行时参数，调度作业时自动计算T-1日期传入即可：

1. 自定义管道选项，添加日期参数：

public interface RetryOptions extends PipelineOptions {
  @Description("待重试失败记录的日期，格式为yyyy/MM/dd")
  ValueProvider<String> getRetryDate();
  void setRetryDate(ValueProvider<String> value);

  @Description("GCS存储桶路径前缀")
  ValueProvider<String> getGcsPathPrefix();
  void setGcsPathPrefix(ValueProvider<String> value);
}

2. 构造动态通配符路径时直接传入ValueProvider，不要提前调用get()方法：

// 拼接带日期的GCS通配符路径
ValueProvider<String> gcsPattern = ValueProvider.NestedValueProvider.of(
  options.getGcsPathPrefix(),
  options.getRetryDate(),
  (prefix, date) -> String.format("%s/%s/*/*.json", prefix, date)
);

// 直接匹配读取文件，去掉冗余的二次match逻辑
PCollection<ReadableFile> targetFiles = input.getPipeline()
  .apply("Match target GCS files", FileIO.match().filepattern(gcsPattern))
  .apply("Convert to readable files", FileIO.readMatches());

// 后续直接读取文件内容、调用重试API即可
targetFiles.apply("Parse failed events", ParDo.of(new ReadFromGCS()))
           .apply("Call retry API", /* 自定义API重试逻辑 */);

3. 调度作业时（可搭配Cloud Scheduler、Cloud Composer使用），每次触发作业前自动计算前一天的日期，作为--retryDate参数传入即可，每次运行作业都会使用最新日期拼接路径，不会出现模板上传时路径被固定的问题。

方案2：DoFn内动态生成路径（无需外部传参） #

如果不想在调度侧维护日期计算逻辑，可以先生成空输入，在DoFn内计算T-1日期、生成目标路径，再通过FileIO.matchAll()完成动态匹配：

// 生成空触发输入
PCollection<Void> trigger = input.getPipeline()
  .apply("Init trigger", Create.of((Void) null));

// 在DoFn内计算目标日期、生成GCS通配符路径
PCollection<String> dynamicPatterns = trigger
  .apply("Generate T-1 GCS path", ParDo.of(new DoFn<Void, String>() {
    @ProcessElement
    public void processElement(OutputReceiver<String> out) {
      String t1Date = Instant.now()
        .minus(Duration.standardDays(1))
        .toDateTime(DateTimeZone.UTC)
        .toString("yyyy/MM/dd");
      out.output(String.format("%s/%s/*/*.json", gcsPathPrefix, t1Date));
    }
  }));

// 动态匹配路径并转换为可读文件
PCollection<ReadableFile> targetFiles = dynamicPatterns
  .apply("Match dynamic paths", FileIO.matchAll())
  .apply("Convert to readable files", FileIO.readMatches());

// 后续接文件解析、API重试逻辑即可

该方案的日期计算逻辑在DoFn内部，只有作业实际运行时才会执行，不会在模板生成阶段被固定。

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