通过谷歌DataFlow转换器查询关系数据库

<dependencies>
  <dependency>
    <groupId>org.apache.beam</groupId>
    <artifactId>beam-sdks-java-io-jdbc</artifactId>
    <version>2.31.0</version>
  </dependency>
  <dependency>
    <groupId>mysql</groupId>
    <artifactId>mysql-connector-java</artifactId>
    <version>8.0.23</version>
  </dependency>
</dependencies>

import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.io.jdbc.JdbcIO;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.ParDo;

public class JdbcQueryExample {

  public static void main(String[] args) {
    PipelineOptions options = PipelineOptionsFactory.fromArgs(args).create();
    Pipeline pipeline = Pipeline.create(options);

    pipeline.apply(JdbcIO.<String>read()
        .withDataSourceConfiguration(JdbcIO.DataSourceConfiguration.create(
            "com.mysql.cj.jdbc.Driver",
            "jdbc:mysql://your_database_host:3306/your_database_name"))
        .withQuery("SELECT * FROM your_table_name")
        .withCoder(StringUtf8Coder.of())
        .apply(ParDo.of(new DoFn<String, Void>() {
          @ProcessElement
          public void processElement(@Element String element, OutputReceiver<Void> out) {
            System.out.println(element);
          }
        })));

    pipeline.run().waitUntilFinish();
  }
}

--project=your_project_id
--region=your_region
--jobName=jdbc_query_example
--runner=DataflowRunner
--tempLocation=gs://your_bucket_name/temp

mvn compile exec:java -Dexec.mainClass="JdbcQueryExample" -Dexec.args="--project=your_project_id --region=your_region --jobName=jdbc_query_example --runner=DataflowRunner --tempLocation=gs://your_bucket_name/temp"