如何使用Apache光束库[org.apache.beam.sdk.io.*]获取/添加GCS文件用户定义的元数据

<dependency>
  <groupId>org.apache.beam</groupId>
  <artifactId>beam-sdks-java-io-google-cloud-platform</artifactId>
  <version>2.30.0</version>
</dependency>

import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.io.gcp.storage.GcsIO;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.ParDo;

public class GCSMetadataExample {
  public static void main(String[] args) {
    // 创建PipelineOptions
    PipelineOptions options = PipelineOptionsFactory.create();

    // 创建Pipeline
    Pipeline pipeline = Pipeline.create(options);

    // 从GCS读取文件元数据
    pipeline
      .apply(GcsIO.readMetadata().from("gs://your-bucket/your-file"))
      .apply(ParDo.of(new ProcessMetadataFn()));

    // 向GCS文件添加用户定义的元数据
    pipeline
      .apply(GcsIO.writeMetadata().to("gs://your-bucket/your-file"))
      .apply(ParDo.of(new AddMetadataFn()));

    // 运行Pipeline
    pipeline.run().waitUntilFinish();
  }

  // 自定义处理元数据的函数
  static class ProcessMetadataFn extends DoFn<GcsIO.ReadableMetadata, Void> {
    @ProcessElement
    public void processElement(ProcessContext c) {
      GcsIO.ReadableMetadata metadata = c.element();
      // 处理元数据，例如打印或存储到数据库
      System.out.println(metadata);
    }
  }

  // 自定义添加元数据的函数
  static class AddMetadataFn extends DoFn<GcsIO.WritableMetadata, Void> {
    @ProcessElement
    public void processElement(ProcessContext c) {
      GcsIO.WritableMetadata metadata = c.element();
      // 添加元数据到描述属性
      metadata.getMetadata().set("key", "value");
      c.output(metadata);
    }
  }
}