io-connectors

用途

I/O 连接器用于从外部数据源读取数据以及向外部数据源写入数据。对于从事 Apache Beam 中 I/O 连接器开发、使用与调试的场景，本技能提供相关指引，包括使用现有连接器、创建新连接器以及调试数据源/数据汇。

主要能力与可用连接器

Beam 提供 51 个以上的 Java I/O 连接器以及若干 Python 连接器。Java I/O 连接器位于 sdks/java/io/。

可用连接器按类别列出如下：

云存储：google-cloud-platform（BigQuery、Bigtable、Spanner、Pub/Sub、GCS）、amazon-web-services2、azure、azure-cosmos
数据库：jdbc、mongodb、cassandra、hbase、redis、neo4j、clickhouse、influxdb、singlestore、elasticsearch
消息：kafka、pulsar、rabbitmq、amqp、jms、mqtt、solace
文件格式：parquet、csv、json、xml、thrift、iceberg
其他：snowflake、splunk、cdap、debezium、hadoop-format、kudu、solr、tika

测试 I/O 连接器

单元测试示例命令：

./gradlew :sdks:java:io:kafka:test
./gradlew :sdks:java:io:jdbc:test

集成测试：

在 Direct Runner 上：

./gradlew :sdks:java:io:google-cloud-platform:integrationTest

使用自定义 GCP 设置：

./gradlew :sdks:java:io:google-cloud-platform:integrationTest \
  -PgcpProject=<project> \
  -PgcpTempRoot=gs://<bucket>/path

使用显式 Pipeline 选项：

./gradlew :sdks:java:io:jdbc:integrationTest \
  -DbeamTestPipelineOptions='["--runner=TestDirectRunner"]'

集成测试框架位于 it/ 目录：

it/common/ — 通用测试工具
it/google-cloud-platform/ — GCP 专用测试基础设施
it/jdbc/ — JDBC 测试基础设施
it/kafka/ — Kafka 测试基础设施
it/testcontainers/ — Testcontainers 支持

编写集成测试的基本结构：

@RunWith(JUnit4.class)
public class MyIOIT {
  @Rule public TestPipeline readPipeline = TestPipeline.create();
  @Rule public TestPipeline writePipeline = TestPipeline.create();

  @Test
  public void testWriteAndRead() {
    // Write data
    writePipeline.apply(Create.of(testData))
                 .apply(MyIO.write().to(destination));
    writePipeline.run().waitUntilFinish();

    // Read and verify
    PCollection<String> results = readPipeline.apply(MyIO.read().from(destination));
    PAssert.that(results).containsInAnyOrder(expectedData);
    readPipeline.run().waitUntilFinish();
  }
}

使用 TestPipeline：

@Rule public TestPipeline pipeline = TestPipeline.create();

TestPipeline 特性：默认在 run 时阻塞（在 TestDataflowRunner 上）、默认 15 分钟超时、从 beamTestPipelineOptions 系统属性读取选项。

使用方式（GCP I/O 连接器示例）

BigQuery：

// Read
pipeline.apply(BigQueryIO.readTableRows().from("project:dataset.table"));

// Write
data.apply(BigQueryIO.writeTableRows()
    .to("project:dataset.table")
    .withSchema(schema)
    .withWriteDisposition(WriteDisposition.WRITE_APPEND));

Pub/Sub：

// Read
pipeline.apply(PubsubIO.readStrings().fromTopic("projects/project/topics/topic"));

// Write
data.apply(PubsubIO.writeStrings().to("projects/project/topics/topic"));

Cloud Storage（TextIO）：

// Read
pipeline.apply(TextIO.read().from("gs://bucket/path/*.txt"));

// Write
data.apply(TextIO.write().to("gs://bucket/output").withSuffix(".txt"));

Kafka 连接器：

// Read
pipeline.apply(KafkaIO.<String, String>read()
    .withBootstrapServers("localhost:9092")
    .withTopic("topic")
    .withKeyDeserializer(StringDeserializer.class)
    .withValueDeserializer(StringDeserializer.class));

// Write
data.apply(KafkaIO.<String, String>write()
    .withBootstrapServers("localhost:9092")
    .withTopic("topic")
    .withKeySerializer(StringSerializer.class)
    .withValueSerializer(StringSerializer.class));

JDBC 连接器：

// Read
pipeline.apply(JdbcIO.<Row>read()
    .withDataSourceConfiguration(JdbcIO.DataSourceConfiguration
        .create("org.postgresql.Driver", "jdbc:postgresql://host/db"))
    .withQuery("SELECT * FROM table"));

// Write
data.apply(JdbcIO.<Row>write()
    .withDataSourceConfiguration(config)
    .withStatement("INSERT INTO table VALUES (?, ?)"));

Python I/O

Python I/O 位于 sdks/python/apache_beam/io/。常用 Python I/O：

textio — 文本文件
fileio — 通用文件操作
avroio — Avro 文件
parquetio — Parquet 文件
gcp/ — GCP 连接器（BigQuery、Pub/Sub、Datastore 等）

跨语言 I/O

Beam 支持通过扩展服务（expansion service）在一个 SDK 中使用另一个 SDK 的 I/O 连接器。启动 Java 扩展服务：

./gradlew :sdks:java:io:expansion-service:runExpansionService

创建新连接器

关键组件：

Source — 读取数据（有界或无界）
Sink — 写入数据
Read/Write transforms — 面向用户的 API

关于开发新 I/O 连接器的更详细信息，请参阅 Developing new I/O connectors SKILL。

适用场景与限制

本技能适用于使用 I/O 连接器、创建新连接器、调试数据源/数据汇，以及编写连接器的单元测试与集成测试。资料未提供具体的平台限制、收费或价格信息。