io-connectors

用途

该技能用于指导 Apache Beam 中 I/O 连接器的开发与使用。当你处理 I/O 连接器、创建新的连接器，或调试数据源/数据汇（data source/sink）时可以使用它。

概述

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 Options：

./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 位置

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 连接器。

# Start Java expansion service
./gradlew :sdks:java:io:expansion-service:runExpansionService

创建新连接器

关键组件：

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

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