io-connectors

Files (2)

• LICENSE
• SKILL.md

SKILL.md (first lines)

# I/O Connectors in Apache Beam

## Overview
I/O connectors enable reading from and writing to external data sources. Beam provides 51+ Java I/O connectors and several Python connectors.

## Java I/O Connectors Location
`sdks/java/io/`

### Available Connectors
| Category | Connectors |
|----------|------------|
| Cloud Storage | google-cloud-platform (BigQuery, Bigtable, Spanner, Pub/Sub, GCS), amazon-web-services2, azure, azure-cosmos |
| Databases | jdbc, mongodb, cassandra, hbase, redis, neo4j, clickhouse, influxdb, singlestore, elasticsearch |
| Messaging | kafka, pulsar, rabbitmq, amqp, jms, mqtt, solace |
| File Formats | parquet, csv, json, xml, thrift, iceberg |
| Other | snowflake, splunk, cdap, debezium, hadoop-format, kudu, solr, tika |

## Testing I/O Connectors

### Unit Tests
```bash
./gradlew :sdks:java:io:kafka:test
./gradlew :sdks:java:io:jdbc:test
```

### Integration Tests

#### On Direct Runner
```bash
./gradlew :sdks:java:io:google-cloud-platform:integrationTest
```

#### With Custom GCP Settings
```bash
./gradlew :sdks:java:io:google-cloud-platform:integrationTest \
  -PgcpProject=<project> \
  -PgcpTempRoot=gs://<bucket>/path
```

#### With Explicit Pipeline Options
```bash
./gradlew :sdks:java:io:jdbc:integrationTest \
  -DbeamTestPipelineOptions='["--runner=TestDirectRunner"]'
```

## Integration Test Framework
Located at `it/` directory:
- `it/common/` - Common test utilities
- `it/google-cloud-platform/` - GCP-specific test infrastructure
- `it/jdbc/` - JDBC test infrastructure
- `it/kafka/` - Kafka test infrastructure
- `it/testcontainers/` - Testcontainers support

## Writing Integration Tests

### Basic Structure
```java
@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();
  }
}
```

### Using TestPipeline
```java
@Rule public TestPipeline pipeline = TestPipeline.create();
```

TestPipeline:
- Blocks on run by default (on TestDataflowRunner)
- Has 15-minute default timeout
- Reads options from `beamTestPipelineOptions` system property

## GCP I/O Connectors

### BigQuery
```java
// 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
```java
// Read
pipeline.apply(PubsubIO.readStrings().fromTopic("projects/project/topics/topic"));

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

### Cloud Storage (TextIO)
```java
// Read
pipeline.apply(TextIO.read().from("gs://bucket/path/*.txt"));

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

From Kilo-Org/kilo-marketplace, Apache-2.0.

More from Kilo-Org/kilo-marketplace