Skip to content

Commit 6f30f14

Browse files
chamikaramjchamikaramj
authored
Adds a new agent SKILL for developing new I/O connectors (#38910)
1 parent 8ea81a9 commit 6f30f14

2 files changed

Lines changed: 347 additions & 0 deletions

File tree

‎.agent/skills/developing-new-io-connectors/SKILL.md‎

Lines changed: 345 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,345 @@
1+
---
2+
# Licensed to the Apache Software Foundation (ASF) under one
3+
# or more contributor license agreements. See the NOTICE file
4+
# distributed with this work for additional information
5+
# regarding copyright ownership. The ASF licenses this file
6+
# to you under the Apache License, Version 2.0 (the
7+
# "License"); you may not use this file except in compliance
8+
# with the License. You may obtain a copy of the License at
9+
#
10+
# http://www.apache.org/licenses/LICENSE-2.0
11+
#
12+
# Unless required by applicable law or agreed to in writing,
13+
# software distributed under the License is distributed on an
14+
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
# KIND, either express or implied. See the License for the
16+
# specific language governing permissions and limitations
17+
# under the License.
18+
19+
name: developing-new-io-connectors
20+
description: End-to-end guide on developing new Apache Beam I/O connectors correctly, including core IO transforms, SchemaTransforms, URN proto definitions, Managed API integration, cross-language expansion service, and testing.
21+
---
22+
23+
# Developing New Apache Beam I/O Connectors
24+
25+
This guide outlines the modern best practices and mandatory steps for building new Apache Beam I/O connectors. Modern Beam connectors are expected to be schema-aware, available in cross-language (Python, Go) pipelines via the Expansion Service, and seamlessly integrable with Beam YAML and the `Managed` I/O API.
26+
27+
## 1. Module Structure and Gradle Setup
28+
29+
New Java I/O connectors should reside under `sdks/java/io/<connector-name>`.
30+
31+
### Directory Layout
32+
```text
33+
sdks/java/io/<connector-name>/
34+
├── build.gradle
35+
└── src/
36+
├── main/java/org/apache/beam/sdk/io/<connector-name>/
37+
│ ├── <Connector>IO.java
38+
│ ├── <Connector>ReadSchemaTransformProvider.java
39+
│ └── <Connector>WriteSchemaTransformProvider.java
40+
└── test/java/org/apache/beam/sdk/io/<connector-name>/
41+
├── <Connector>IOTest.java
42+
└── <Connector>ReadSchemaTransformProviderTest.java
43+
```
44+
45+
### Gradle Configuration (`build.gradle`)
46+
Your `build.gradle` must use standard Beam Java module conventions and explicitly declare necessary dependencies.
47+
48+
```groovy
49+
plugins { id 'org.apache.beam.module' }
50+
applyJavaNature(
51+
// If <connector-name> contains hyphens, convert them to dots or underscores
52+
automaticModuleName: 'org.apache.beam.sdk.io.<connector_name>',
53+
)
54+
55+
description = "Apache Beam :: SDKs :: Java :: IO :: <Connector Name>"
56+
ext.summary = "Integration with <External System>."
57+
58+
dependencies {
59+
implementation project(path: ":sdks:java:core", configuration: "shadow")
60+
implementation project(path: ":model:pipeline", configuration: "shadow") // For URN definitions
61+
62+
// Add external client libraries here
63+
implementation library.java.<client_dependency>
64+
65+
// Handle strict dependency checking if necessary
66+
permitUnusedDeclared library.java.<client_dependency>
67+
68+
// Standard test dependencies
69+
testImplementation project(path: ":sdks:java:core", configuration: "shadowTest")
70+
testImplementation library.java.junit
71+
}
72+
```
73+
74+
---
75+
76+
## 2. Core I/O Transform Implementation (`<Connector>IO.java`)
77+
78+
Follow Beam's canonical AutoValue builder pattern for user-facing API configuration. While core Java I/O connectors can be strongly typed using specific domain classes or Java generics (`<T>`) for idiomatic Java SDK usage, modern sources should also emphasize Beam `Row` and Schema support (e.g., via `.readRows()`). For excellent real-world implementations of this pattern, refer to `IcebergIO` and `DeltaIO`.
79+
80+
### Bounded & Unbounded Sources
81+
Instead of legacy `Source` classes, implement reading via Beam's **Splittable DoFns (SDF)** framework for advanced features such as dynamic rebalancing and watermark support.
82+
83+
A primary read transform (such as `read()` or `readRows()`) typically extends `PTransform<PBegin, PCollection<T>>` (or `PCollection<Row>`). Using `PCollection` as input is meant for "ReadAll" operations (such as reading a collection of file patterns or queries).
84+
85+
An example SDF-based read transform is given below:
86+
87+
```java
88+
public class MyIO {
89+
public static ReadRows readRows() {
90+
return new AutoValue_MyIO_ReadRows.Builder().build();
91+
}
92+
93+
@AutoValue
94+
public abstract static class ReadRows extends PTransform<PBegin, PCollection<Row>> {
95+
public abstract @Nullable String getConfigurationOption();
96+
public abstract @Nullable Schema getSchema();
97+
public abstract Builder toBuilder();
98+
99+
@AutoValue.Builder
100+
public abstract static class Builder {
101+
public abstract Builder setConfigurationOption(String value);
102+
public abstract Builder setSchema(Schema schema);
103+
public abstract ReadRows build();
104+
}
105+
106+
public ReadRows withConfigurationOption(String value) {
107+
return toBuilder().setConfigurationOption(value).build();
108+
}
109+
110+
public ReadRows withSchema(Schema schema) {
111+
return toBuilder().setSchema(schema).build();
112+
}
113+
114+
@Override
115+
public PCollection<Row> expand(PBegin input) {
116+
return input
117+
// `ReaderDoFn` is an SDF or source implementation that reads records and outputs `Row` objects.
118+
.apply(ParDo.of(new ReaderDoFn(getConfigurationOption())))
119+
.setRowSchema(getSchema());
120+
}
121+
}
122+
123+
public static WriteRows writeRows() {
124+
return new AutoValue_MyIO_WriteRows.Builder().build();
125+
}
126+
127+
@AutoValue
128+
public abstract static class WriteRows extends PTransform<PCollection<Row>, PDone> {
129+
public abstract @Nullable String getConfigurationOption();
130+
public abstract Builder toBuilder();
131+
132+
@AutoValue.Builder
133+
public abstract static class Builder {
134+
public abstract Builder setConfigurationOption(String value);
135+
public abstract WriteRows build();
136+
}
137+
138+
public WriteRows withConfigurationOption(String value) {
139+
return toBuilder().setConfigurationOption(value).build();
140+
}
141+
142+
@Override
143+
public PDone expand(PCollection<Row> input) {
144+
input.apply("WriteRecords", ParDo.of(new WriterDoFn(getConfigurationOption())));
145+
return PDone.in(input.getPipeline());
146+
}
147+
}
148+
}
149+
```
150+
* Please make sure that any code you add adheres to the Beam coding standards. These standards are documented here: https://beam.apache.org/contribute/code-guidelines/
151+
152+
* Especially scrutinize any logic that involves splitting the data for parallel processing since it is a common source of errors that can lead to data loss or data duplication related issues.
153+
154+
### Developing Sinks (Write Transforms)
155+
When implementing data egress (Write transforms), avoid creating single-worker bottlenecks. Depending on your target system's transactional requirements, prefer one of the following canonical Beam sink patterns:
156+
157+
1. **DoFn-Based / Batching Sinks:** For external APIs or messaging systems (e.g., Kafka, Pub/Sub, NoSQL databases), use a `DoFn` that manages connections per bundle (`@StartBundle`, `@FinishBundle`) or utilizes `GroupIntoBatches` to perform highly efficient, parallel batched requests.
158+
2. **Two-Phase Commit / Exactly-Once Sinks:** For transactional sinks (e.g., Relational DBs, Apache Iceberg, Delta Lake), implement a multi-stage `PTransform`:
159+
* **Write Shards:** Write records in parallel tasks to staging files or temporary transactions, emitting commit descriptors (`PCollection<CommitMessage>`).
160+
* **Global Commit:** Group the commit messages and execute a single final transaction to commit all staged files/shards.
161+
3. **File-Based Sinks:** If your connector purely writes files, leverage Beam's `FileIO` core infrastructure (`FileIO.write()` / `FileIO.Sink`) rather than implementing custom file rolling and sharding logic.
162+
4. **Error Reporting (Dead-Letter Queues):** Instead of failing the entire pipeline when an invalid element or API error occurs, modern Write transforms should optionally output a `PCollection<Row>` (or custom `WriteResult` / `PCollectionRowTuple`) containing failed records and explicit error metadata.
163+
164+
* Specially scrutinize any logic that can create duplicate data due to worker failures. Assume that any transform in the pipeline can fail and be retried multiple times by the Beam runner. If the sink does not handle this properly, it can lead to duplicate data in the target system.
165+
---
166+
167+
## 3. Extending the Pipeline Model Proto (`external_transforms.proto`)
168+
169+
To standardize your transform identifier across SDKs, define its URN in Beam's protobuf schema.
170+
171+
1. Open `model/pipeline/src/main/proto/org/apache/beam/model/pipeline/v1/external_transforms.proto`.
172+
2. Add your read/write URNs under the appropriate enum (e.g., `ManagedTransforms.Urns`).
173+
174+
```protobuf
175+
message ManagedTransforms {
176+
enum Urns {
177+
// ... existing entries
178+
MY_SYSTEM_READ = 15 [(org.apache.beam.model.pipeline.v1.beam_urn) =
179+
"beam:schematransform:org.apache.beam:my_system_read:v1"];
180+
MY_SYSTEM_WRITE = 16 [(org.apache.beam.model.pipeline.v1.beam_urn) =
181+
"beam:schematransform:org.apache.beam:my_system_write:v1"];
182+
}
183+
}
184+
```
185+
186+
3. Re-generate and compile the model protos:
187+
```bash
188+
./gradlew :model:pipeline:generateProto :model:pipeline:compileJava
189+
```
190+
191+
---
192+
193+
## 4. Implementing `SchemaTransformProvider`
194+
195+
To expose your connector to cross-language pipelines and Beam YAML, create a typed `SchemaTransformProvider`.
196+
197+
```java
198+
@AutoService(SchemaTransformProvider.class)
199+
public class MyReadSchemaTransformProvider extends TypedSchemaTransformProvider<Configuration> {
200+
201+
@Override
202+
public String identifier() {
203+
return getUrn(ExternalTransforms.ManagedTransforms.Urns.MY_SYSTEM_READ);
204+
}
205+
206+
@Override
207+
public String description() {
208+
return "Reads records from My System and outputs a PCollection of Beam Rows.";
209+
}
210+
211+
@Override
212+
protected SchemaTransform from(Configuration configuration) {
213+
return new MyReadSchemaTransform(configuration);
214+
}
215+
216+
@Override
217+
public List<String> outputCollectionNames() {
218+
return Collections.singletonList("output");
219+
}
220+
221+
@DefaultSchema(AutoValueSchema.class)
222+
@AutoValue
223+
public abstract static class Configuration {
224+
@SchemaFieldDescription("Configuration option description.")
225+
public abstract String getConfigurationOption();
226+
227+
public static Builder builder() {
228+
return new AutoValue_MyReadSchemaTransformProvider_Configuration.Builder();
229+
}
230+
231+
@AutoValue.Builder
232+
public abstract static class Builder {
233+
public abstract Builder setConfigurationOption(String value);
234+
public abstract Configuration build();
235+
}
236+
}
237+
238+
static class MyReadSchemaTransform extends SchemaTransform {
239+
private final Configuration configuration;
240+
241+
MyReadSchemaTransform(Configuration configuration) {
242+
this.configuration = Objects.requireNonNull(configuration, "configuration cannot be null");
243+
}
244+
245+
@Override
246+
public PCollectionRowTuple expand(PCollectionRowTuple input) {
247+
PCollection<Row> output = input.getPipeline().apply(
248+
MyIO.readRows().withConfigurationOption(configuration.getConfigurationOption()));
249+
return PCollectionRowTuple.of("output", output);
250+
}
251+
}
252+
}
253+
```
254+
255+
---
256+
257+
## 5. Integrating with Managed API (`Managed.java`)
258+
259+
Beam's `Managed` I/O transform provides a unified interface for data ingest/egress. To support it:
260+
261+
1. Open `sdks/java/managed/src/main/java/org/apache/beam/sdk/managed/Managed.java`.
262+
2. Define a public constant identifier:
263+
```java
264+
public static final String MY_SYSTEM = "my_system";
265+
```
266+
3. Register your URNs in `READ_TRANSFORMS` or `WRITE_TRANSFORMS`:
267+
```java
268+
public static final Map<String, String> READ_TRANSFORMS =
269+
ImmutableMap.<String, String>builder()
270+
// ... existing transforms
271+
.put(MY_SYSTEM, getUrn(ExternalTransforms.ManagedTransforms.Urns.MY_SYSTEM_READ))
272+
.build();
273+
```
274+
4. Update the Javadoc block in `Managed.java` to list your new connector.
275+
276+
---
277+
278+
## 6. Expansion Service Registration
279+
280+
To enable non-Java SDKs (Python, Go) to discover and expand your new connector, include it in the standard Java Expansion Service.
281+
282+
1. Open `sdks/java/io/expansion-service/build.gradle`.
283+
2. Add your module as a runtime dependency:
284+
```groovy
285+
dependencies {
286+
// ... existing dependencies
287+
runtimeOnly project(":sdks:java:io:<connector-name>")
288+
}
289+
```
290+
291+
---
292+
293+
## 7. Python & Beam YAML Integration
294+
295+
Once registered in the expansion service, your `SchemaTransform` can be utilized in Python and YAML. E.g., for `Managed` support in Python:
296+
297+
1. Open `sdks/python/apache_beam/transforms/managed.py`.
298+
2. Export your identifier in `__all__` and map it to its URN in `Read._READ_TRANSFORMS` or `Write._WRITE_TRANSFORMS`:
299+
```python
300+
MY_SYSTEM = 'my_system'
301+
302+
__all__ = [
303+
# ... existing
304+
"MY_SYSTEM",
305+
]
306+
307+
class Read(PTransform):
308+
_READ_TRANSFORMS = {
309+
# ... existing
310+
MY_SYSTEM: ManagedTransforms.Urns.MY_SYSTEM_READ.urn,
311+
}
312+
```
313+
3. In `sdks/python/apache_beam/transforms/external.py`, map the URN to the appropriate Expansion Service jar target in `MANAGED_TRANSFORM_URN_TO_JAR_TARGET_MAPPING`.
314+
315+
---
316+
317+
## 8. Verification and Testing
318+
319+
Verify your new connector thoroughly across multiple abstraction layers:
320+
321+
### 1. Unit Tests
322+
Test your core builder and `SchemaTransformProvider` translation.
323+
```bash
324+
./gradlew :sdks:java:io:<connector-name>:test
325+
```
326+
327+
### 2. Managed API Translation
328+
In `ManagedSchemaTransformTranslationTest.java` (under `sdks/java/managed`), you can verify the translation structure of your managed transform. Note that `ManagedTest.java` generally uses dummy/test providers (`TestSchemaTransformProvider`) to keep dependencies lightweight.
329+
```bash
330+
./gradlew :sdks:java:managed:test
331+
```
332+
333+
### 3. Integration Tests (IT)
334+
Create integration tests to test end-to-end data processing against real system instances, including `Managed.read(Managed.MY_SYSTEM)` usage.
335+
336+
* Test resources should be managed via `ResourceManager` classes under the `it/` directory.
337+
* Add GitHub Actions to trigger your tests when changes are made to your connector code. Consider adding one for pre-commit and one for post-commit.
338+
339+
### 4. Documentation
340+
Add any necessary documentation for your connector under the `website/www/site/content/en/documentation/io/built-in/` directory.
341+
342+
---
343+
344+
> [!TIP]
345+
> **Canonical Reference Implementations:** When developing a new connector, we highly recommend studying **Apache Iceberg** ([IcebergIO.java](https://github.com/apache/beam/blob/master/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/IcebergIO.java)) and **Delta Lake** ([DeltaIO.java](https://github.com/apache/beam/blob/master/sdks/java/io/delta/src/main/java/org/apache/beam/sdk/io/delta/DeltaIO.java)) as state-of-the-art reference implementations.

‎.agent/skills/io-connectors/SKILL.md‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -195,3 +195,5 @@ Key components:
195195
1. **Source** - Reads data (bounded or unbounded)
196196
2. **Sink** - Writes data
197197
3. **Read/Write transforms** - User-facing API
198+
199+
For more detailed information see the [Developing I/O connectors](https://beam.apache.org/documentation/io/developing-io-overview) guide.

0 commit comments

Comments
 (0)