You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by "chamikaramj (via GitHub)" <gi...@apache.org> on 2026/06/10 23:28:12 UTC
[PR] Adds a new agent SKILL for developing new I/O connectors [beam]
chamikaramj opened a new pull request, #38910:
URL: https://github.com/apache/beam/pull/38910
Covers basic instructions for developing new sources and sinks and scrutinizes several cases that can lead to bugs.
------------------------
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
- [ ] Mention the appropriate issue in your description (for example: `addresses #123`), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, comment `fixes #<ISSUE NUMBER>` instead.
- [ ] Update `CHANGES.md` with noteworthy changes.
- [ ] If this contribution is large, please file an Apache [Individual Contributor License Agreement](https://www.apache.org/licenses/icla.pdf).
See the [Contributor Guide](https://beam.apache.org/contribute) for more tips on [how to make review process smoother](https://github.com/apache/beam/blob/master/CONTRIBUTING.md#make-the-reviewers-job-easier).
To check the build health, please visit [https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md](https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md)
GitHub Actions Tests Status (on master branch)
------------------------------------------------------------------------------------------------
[](https://github.com/apache/beam/actions?query=workflow%3A%22Build+python+source+distribution+and+wheels%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Python+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Java+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Go+tests%22+branch%3Amaster+event%3Aschedule)
See [CI.md](https://github.com/apache/beam/blob/master/CI.md) for more information about GitHub Actions CI or the [workflows README](https://github.com/apache/beam/blob/master/.github/workflows/README.md) to see a list of phrases to trigger workflows.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Adds a new agent SKILL for developing new I/O connectors [beam]
Posted by "chamikaramj (via GitHub)" <gi...@apache.org>.
chamikaramj merged PR #38910:
URL: https://github.com/apache/beam/pull/38910
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Adds a new agent SKILL for developing new I/O connectors [beam]
Posted by "gemini-code-assist[bot] (via GitHub)" <gi...@apache.org>.
gemini-code-assist[bot] commented on PR #38910:
URL: https://github.com/apache/beam/pull/38910#issuecomment-4675671852
## Summary of Changes
Hello, I'm Gemini Code Assist[^1]! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed!
This pull request introduces a detailed guide for developers to build new Apache Beam I/O connectors. It standardizes the development process by providing best practices for module structure, transform implementation, and cross-language integration, ensuring new connectors are robust, schema-aware, and easily accessible across different SDKs.
### Highlights
* **New I/O Connector Development Guide**: Added a comprehensive guide for developing new Apache Beam I/O connectors, covering module structure, Gradle configuration, and implementation patterns.
* **Modern Best Practices**: Outlined requirements for schema-aware connectors, cross-language support via the Expansion Service, and integration with the Managed I/O API.
* **Documentation Updates**: Updated existing I/O connector documentation to include a reference link to the new development guide.
<details>
<summary><b>New Features</b></summary>
🧠You can now enable **Memory** (public preview) to help **Gemini Code Assist** learn from your team's feedback. This makes future code reviews more consistent and personalized to your project's style. **Click [here](https://codeassist.google/code-review/login) to enable Memory in your admin console.**
</details>
<details>
<summary><b>Using Gemini Code Assist</b></summary>
<br>
The full guide for Gemini Code Assist can be found on our [documentation page](https://developers.google.com/gemini-code-assist/docs/review-github-code), here are some quick tips.
<b>Invoking Gemini</b>
You can request assistance from Gemini at any point by creating a comment using either `/gemini <command>` or `@gemini-code-assist <command>`. Below is a summary of the supported commands on the current page.
Feature | Command | Description
--- | --- | ---
Code Review | `/gemini review` | Performs a code review for the current pull request in its current state.
Pull Request Summary | `/gemini summary` | Provides a summary of the current pull request in its current state.
Comment | @gemini-code-assist | Responds in comments when explicitly tagged, both in pull request comments and review comments.
Help | `/gemini help` | Displays a list of available commands.
<b>Customization</b>
To customize the Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a `.gemini/` folder in the base of the repository. Detailed instructions can be found [here](https://developers.google.com/gemini-code-assist/docs/customize-gemini-behavior-github).
<b>Limitations & Feedback</b>
Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counterproductive. You can react with :thumbsup: and :thumbsdown: on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for GitHub and other Google products, sign up [here](https://google.qualtrics.com/jfe/form/SV_2cyuGuTWsEw84yG).
</details>
[^1]: Review the [Privacy Notices](https://policies.google.com/privacy), [Generative AI Prohibited Use Policy](https://policies.google.com/terms/generative-ai/use-policy), [Terms of Service](https://policies.google.com/terms), and learn how to configure Gemini Code Assist in GitHub [here](https://developers.google.com/gemini-code-assist/docs/customize-gemini-behavior-github). Gemini can make mistakes, so double check it and [use code with caution](https://support.google.com/legal/answer/13505487).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Adds a new agent SKILL for developing new I/O connectors [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #38910:
URL: https://github.com/apache/beam/pull/38910#issuecomment-4682100885
Assigning reviewers:
R: @tvalentyn added as fallback since no labels match configuration
Note: If you would like to opt out of this review, comment `assign to next reviewer`.
Available commands:
- `stop reviewer notifications` - opt out of the automated review tooling
- `remind me after tests pass` - tag the comment author after tests pass
- `waiting on author` - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)
The PR bot will only process comments in the main thread (not review comments).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Adds a new agent SKILL for developing new I/O connectors [beam]
Posted by "chamikaramj (via GitHub)" <gi...@apache.org>.
chamikaramj commented on PR #38910:
URL: https://github.com/apache/beam/pull/38910#issuecomment-4682319961
Thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Adds a new agent SKILL for developing new I/O connectors [beam]
Posted by "gemini-code-assist[bot] (via GitHub)" <gi...@apache.org>.
gemini-code-assist[bot] commented on code in PR #38910:
URL: https://github.com/apache/beam/pull/38910#discussion_r3392233578
##########
.agent/skills/developing-new-io-connectors/SKILL.md:
##########
@@ -0,0 +1,345 @@
+---
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+name: developing-new-io-connectors
+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.
+---
+
+# Developing New Apache Beam I/O Connectors
+
+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.
+
+## 1. Module Structure and Gradle Setup
+
+New Java I/O connectors should reside under `sdks/java/io/<connector-name>`.
+
+### Directory Layout
+```text
+sdks/java/io/<connector-name>/
+├── build.gradle
+└── src/
+ ├── main/java/org/apache/beam/sdk/io/<connector-name>/
+ │ ├── <Connector>IO.java
+ │ ├── <Connector>ReadSchemaTransformProvider.java
+ │ └── <Connector>WriteSchemaTransformProvider.java
+ └── test/java/org/apache/beam/sdk/io/<connector-name>/
+ ├── <Connector>IOTest.java
+ └── <Connector>ReadSchemaTransformProviderTest.java
+```
+
+### Gradle Configuration (`build.gradle`)
+Your `build.gradle` must use standard Beam Java module conventions and explicitly declare necessary dependencies.
+
+```groovy
+plugins { id 'org.apache.beam.module' }
+applyJavaNature(
+ // If <connector-name> contains hyphens, convert them to dots or underscores
+ automaticModuleName: 'org.apache.beam.sdk.io.<connector_name>',
+)
+
+description = "Apache Beam :: SDKs :: Java :: IO :: <Connector Name>"
+ext.summary = "Integration with <External System>."
+
+dependencies {
+ implementation project(path: ":sdks:java:core", configuration: "shadow")
+ implementation project(path: ":model:pipeline", configuration: "shadow") // For URN definitions
+
+ // Add external client libraries here
+ implementation library.java.<client_dependency>
+
+ // Handle strict dependency checking if necessary
+ permitUnusedDeclared library.java.<client_dependency>
+
+ // Standard test dependencies
+ testImplementation project(path: ":sdks:java:core", configuration: "shadowTest")
+ testImplementation library.java.junit
+}
+```
+
+---
+
+## 2. Core I/O Transform Implementation (`<Connector>IO.java`)
+
+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`.
+
+### Bounded & Unbounded Sources
+Instead of legacy `Source` classes, implement reading via Beam's **Splittable DoFns (SDF)** framework for advanced features such as dynamic rebalancing and watermark support.
+
+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).
+
+An example SDF-based read transform is given below:
+
+```java
+public class MyIO {
+ public static ReadRows readRows() {
+ return new AutoValue_MyIO_ReadRows.Builder().build();
+ }
+
+ @AutoValue
+ public abstract static class ReadRows extends PTransform<PBegin, PCollection<Row>> {
+ public abstract @Nullable String getConfigurationOption();
+ public abstract @Nullable Schema getSchema();
+ public abstract Builder toBuilder();
+
+ @AutoValue.Builder
+ public abstract static class Builder {
+ public abstract Builder setConfigurationOption(String value);
+ public abstract Builder setSchema(Schema schema);
+ public abstract ReadRows build();
+ }
+
+ public ReadRows withConfigurationOption(String value) {
+ return toBuilder().setConfigurationOption(value).build();
+ }
+
+ public ReadRows withSchema(Schema schema) {
+ return toBuilder().setSchema(schema).build();
+ }
+
+ @Override
+ public PCollection<Row> expand(PBegin input) {
+ return input
+ // `ReaderDoFn` is an SDF or source implementation that reads records and outputs `Row` objects.
+ .apply(ParDo.of(new ReaderDoFn(getConfigurationOption())))
+ .setRowSchema(getSchema());
+ }
+ }
+
+ public static WriteRows writeRows() {
+ return new AutoValue_MyIO_WriteRows.Builder().build();
+ }
+
+ @AutoValue
+ public abstract static class WriteRows extends PTransform<PCollection<Row>, PDone> {
+ public abstract @Nullable String getConfigurationOption();
+ public abstract Builder toBuilder();
+
+ @AutoValue.Builder
+ public abstract static class Builder {
+ public abstract Builder setConfigurationOption(String value);
+ public abstract WriteRows build();
+ }
+
+ public WriteRows withConfigurationOption(String value) {
+ return toBuilder().setConfigurationOption(value).build();
+ }
+
+ @Override
+ public PDone expand(PCollection<Row> input) {
+ input.apply("WriteRecords", ParDo.of(new WriterDoFn(getConfigurationOption())));
+ return PDone.in(input.getPipeline());
+ }
+ }
+}
+```
+* 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/
+
+* 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.
+
+### Developing Sinks (Write Transforms)
+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:
+
+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.
+2. **Two-Phase Commit / Exactly-Once Sinks:** For transactional sinks (e.g., Relational DBs, Apache Iceberg, Delta Lake), implement a multi-stage `PTransform`:
+ * **Write Shards:** Write records in parallel tasks to staging files or temporary transactions, emitting commit descriptors (`PCollection<CommitMessage>`).
+ * **Global Commit:** Group the commit messages and execute a single final transaction to commit all staged files/shards.
+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.
+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.
+
+* 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.
+---
+
+## 3. Extending the Pipeline Model Proto (`external_transforms.proto`)
+
+To standardize your transform identifier across SDKs, define its URN in Beam's protobuf schema.
+
+1. Open `model/pipeline/src/main/proto/org/apache/beam/model/pipeline/v1/external_transforms.proto`.
+2. Add your read/write URNs under the appropriate enum (e.g., `ManagedTransforms.Urns`).
+
+```protobuf
+message ManagedTransforms {
+ enum Urns {
+ // ... existing entries
+ MY_SYSTEM_READ = 15 [(org.apache.beam.model.pipeline.v1.beam_urn) =
+ "beam:schematransform:org.apache.beam:my_system_read:v1"];
+ MY_SYSTEM_WRITE = 16 [(org.apache.beam.model.pipeline.v1.beam_urn) =
+ "beam:schematransform:org.apache.beam:my_system_write:v1"];
+ }
+}
+```
+
+3. Re-generate and compile the model protos:
+```bash
+./gradlew :model:pipeline:generateProto :model:pipeline:compileJava
+```
+
+---
+
+## 4. Implementing `SchemaTransformProvider`
+
+To expose your connector to cross-language pipelines and Beam YAML, create a typed `SchemaTransformProvider`.
+
+```java
+@AutoService(SchemaTransformProvider.class)
+public class MyReadSchemaTransformProvider extends TypedSchemaTransformProvider<Configuration> {
+
+ @Override
+ public String identifier() {
+ return getUrn(ExternalTransforms.ManagedTransforms.Urns.MY_SYSTEM_READ);
+ }
+
+ @Override
+ public String description() {
+ return "Reads records from My System and outputs a PCollection of Beam Rows.";
+ }
+
+ @Override
+ protected SchemaTransform from(Configuration configuration) {
+ return new MyReadSchemaTransform(configuration);
+ }
+
+ @Override
+ public List<String> outputCollectionNames() {
+ return Collections.singletonList("output");
+ }
+
+ @DefaultSchema(AutoValueSchema.class)
+ @AutoValue
+ public abstract static class Configuration {
+ @SchemaFieldDescription("Configuration option description.")
+ public abstract String getConfigurationOption();
+
+ public static Builder builder() {
+ return new AutoValue_MyReadSchemaTransformProvider_Configuration.Builder();
+ }
+
+ @AutoValue.Builder
+ public abstract static class Builder {
+ public abstract Builder setConfigurationOption(String value);
+ public abstract Configuration build();
+ }
+ }
+
+ static class MyReadSchemaTransform extends SchemaTransform {
+ private final Configuration configuration;
+
+ MyReadSchemaTransform(Configuration configuration) {
+ this.configuration = Objects.requireNonNull(configuration, "configuration cannot be null");
+ }
+
+ @Override
+ public PCollectionRowTuple expand(PCollectionRowTuple input) {
+ PCollection<Row> output = input.getPipeline().apply(
+ MyIO.readRows().withConfigurationOption(configuration.getConfigurationOption()));
+ return PCollectionRowTuple.of("output", output);
+ }
+ }
+}
+```
+
+---
+
+## 5. Integrating with Managed API (`Managed.java`)
+
+Beam's `Managed` I/O transform provides a unified interface for data ingest/egress. To support it:
+
+1. Open `sdks/java/managed/src/main/java/org/apache/beam/sdk/managed/Managed.java`.
+2. Define a public constant identifier:
+```java
+public static final String MY_SYSTEM = "my_system";
+```
+3. Register your URNs in `READ_TRANSFORMS` or `WRITE_TRANSFORMS`:
+```java
+public static final Map<String, String> READ_TRANSFORMS =
+ ImmutableMap.<String, String>builder()
+ // ... existing transforms
+ .put(MY_SYSTEM, getUrn(ExternalTransforms.ManagedTransforms.Urns.MY_SYSTEM_READ))
+ .build();
+```
+4. Update the Javadoc block in `Managed.java` to list your new connector.
+
+---
+
+## 6. Expansion Service Registration
+
+To enable non-Java SDKs (Python, Go) to discover and expand your new connector, include it in the standard Java Expansion Service.
+
+1. Open `sdks/java/io/expansion-service/build.gradle`.
+2. Add your module as a runtime dependency:
+```groovy
+dependencies {
+ // ... existing dependencies
+ runtimeOnly project(":sdks:java:io:<connector-name>")
+}
+```
+
+---
+
+## 7. Python & Beam YAML Integration
+
+Once registered in the expansion service, your `SchemaTransform` can be utilized in Python and YAML. E.g., for `Managed` support in Python:
+
+1. Open `sdks/python/apache_beam/transforms/managed.py`.
+2. Export your identifier in `__all__` and map it to its URN in `Read._READ_TRANSFORMS` or `Write._WRITE_TRANSFORMS`:
+```python
+MY_SYSTEM = 'my_system'
+
+__all__ = [
+ # ... existing
+ "MY_SYSTEM",
+]
+
+class Read(PTransform):
+ _READ_TRANSFORMS = {
+ # ... existing
+ MY_SYSTEM: ManagedTransforms.Urns.MY_SYSTEM_READ.urn,
+ }
+```
+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`.
+
+---
+
+## 8. Verification and Testing
+
+Verify your new connector thoroughly across multiple abstraction layers:
+
+### 1. Unit Tests
+Test your core builder and `SchemaTransformProvider` translation.
+```bash
+./gradlew :sdks:java:io:<connector-name>:test
+```
+
+### 2. Managed API Translation
+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.
+```bash
+./gradlew :sdks:java:managed:test
+```
+
+### 3. Integration Tests (IT)
+Create integration tests to test end-to-end data processing against real system instances, including `Managed.read(Managed.MY_SYSTEM)` usage.
+
+* Test resources should be managed via `ResourceManager` classes under the `it/` directory.
+* 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.
+
+### 4. Documentation
+Add any necessary documentation for your connector under the `http://website/www/site/content/en/documentation/io/built-in/` directory.
Review Comment:
The directory path is prefixed with `http://website/`, which makes it look like an invalid URL instead of a repository path. It should be a relative repository path.
```suggestion
Add any necessary documentation for your connector under the `website/www/site/content/en/documentation/io/built-in/` directory.
```
##########
.agent/skills/developing-new-io-connectors/SKILL.md:
##########
@@ -0,0 +1,345 @@
+---
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+name: developing-new-io-connectors
+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.
+---
+
+# Developing New Apache Beam I/O Connectors
+
+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.
+
+## 1. Module Structure and Gradle Setup
+
+New Java I/O connectors should reside under `sdks/java/io/<connector-name>`.
+
+### Directory Layout
+```text
+sdks/java/io/<connector-name>/
+├── build.gradle
+└── src/
+ ├── main/java/org/apache/beam/sdk/io/<connector-name>/
+ │ ├── <Connector>IO.java
+ │ ├── <Connector>ReadSchemaTransformProvider.java
+ │ └── <Connector>WriteSchemaTransformProvider.java
+ └── test/java/org/apache/beam/sdk/io/<connector-name>/
+ ├── <Connector>IOTest.java
+ └── <Connector>ReadSchemaTransformProviderTest.java
+```
+
+### Gradle Configuration (`build.gradle`)
+Your `build.gradle` must use standard Beam Java module conventions and explicitly declare necessary dependencies.
+
+```groovy
+plugins { id 'org.apache.beam.module' }
+applyJavaNature(
+ // If <connector-name> contains hyphens, convert them to dots or underscores
+ automaticModuleName: 'org.apache.beam.sdk.io.<connector_name>',
+)
+
+description = "Apache Beam :: SDKs :: Java :: IO :: <Connector Name>"
+ext.summary = "Integration with <External System>."
+
+dependencies {
+ implementation project(path: ":sdks:java:core", configuration: "shadow")
+ implementation project(path: ":model:pipeline", configuration: "shadow") // For URN definitions
+
+ // Add external client libraries here
+ implementation library.java.<client_dependency>
+
+ // Handle strict dependency checking if necessary
+ permitUnusedDeclared library.java.<client_dependency>
+
+ // Standard test dependencies
+ testImplementation project(path: ":sdks:java:core", configuration: "shadowTest")
+ testImplementation library.java.junit
+}
+```
+
+---
+
+## 2. Core I/O Transform Implementation (`<Connector>IO.java`)
+
+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`.
+
+### Bounded & Unbounded Sources
+Instead of legacy `Source` classes, implement reading via Beam's **Splittable DoFns (SDF)** framework for advanced features such as dynamic rebalancing and watermark support.
+
+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).
+
+An example SDF-based read transform is given below:
+
+```java
+public class MyIO {
+ public static ReadRows readRows() {
+ return new AutoValue_MyIO_ReadRows.Builder().build();
+ }
+
+ @AutoValue
+ public abstract static class ReadRows extends PTransform<PBegin, PCollection<Row>> {
+ public abstract @Nullable String getConfigurationOption();
+ public abstract @Nullable Schema getSchema();
+ public abstract Builder toBuilder();
+
+ @AutoValue.Builder
+ public abstract static class Builder {
+ public abstract Builder setConfigurationOption(String value);
+ public abstract Builder setSchema(Schema schema);
+ public abstract ReadRows build();
+ }
+
+ public ReadRows withConfigurationOption(String value) {
+ return toBuilder().setConfigurationOption(value).build();
+ }
+
+ public ReadRows withSchema(Schema schema) {
+ return toBuilder().setSchema(schema).build();
+ }
+
+ @Override
+ public PCollection<Row> expand(PBegin input) {
+ return input
+ // `ReaderDoFn` is an SDF or source implementation that reads records and outputs `Row` objects.
+ .apply(ParDo.of(new ReaderDoFn(getConfigurationOption())))
+ .setRowSchema(getSchema());
+ }
+ }
+
+ public static WriteRows writeRows() {
+ return new AutoValue_MyIO_WriteRows.Builder().build();
+ }
+
+ @AutoValue
+ public abstract static class WriteRows extends PTransform<PCollection<Row>, PDone> {
+ public abstract @Nullable String getConfigurationOption();
+ public abstract Builder toBuilder();
+
+ @AutoValue.Builder
+ public abstract static class Builder {
+ public abstract Builder setConfigurationOption(String value);
+ public abstract WriteRows build();
+ }
+
+ public WriteRows withConfigurationOption(String value) {
+ return toBuilder().setConfigurationOption(value).build();
+ }
+
+ @Override
+ public PDone expand(PCollection<Row> input) {
+ input.apply("WriteRecords", ParDo.of(new WriterDoFn(getConfigurationOption())));
+ return PDone.in(input.getPipeline());
+ }
+ }
+}
+```
+* 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/
+
+* 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.
+
+### Developing Sinks (Write Transforms)
+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:
+
+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.
+2. **Two-Phase Commit / Exactly-Once Sinks:** For transactional sinks (e.g., Relational DBs, Apache Iceberg, Delta Lake), implement a multi-stage `PTransform`:
+ * **Write Shards:** Write records in parallel tasks to staging files or temporary transactions, emitting commit descriptors (`PCollection<CommitMessage>`).
+ * **Global Commit:** Group the commit messages and execute a single final transaction to commit all staged files/shards.
+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.
+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.
+
+* 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.
+---
+
+## 3. Extending the Pipeline Model Proto (`external_transforms.proto`)
+
+To standardize your transform identifier across SDKs, define its URN in Beam's protobuf schema.
+
+1. Open `model/pipeline/src/main/proto/org/apache/beam/model/pipeline/v1/external_transforms.proto`.
+2. Add your read/write URNs under the appropriate enum (e.g., `ManagedTransforms.Urns`).
+
+```protobuf
+message ManagedTransforms {
+ enum Urns {
+ // ... existing entries
+ MY_SYSTEM_READ = 15 [(org.apache.beam.model.pipeline.v1.beam_urn) =
+ "beam:schematransform:org.apache.beam:my_system_read:v1"];
+ MY_SYSTEM_WRITE = 16 [(org.apache.beam.model.pipeline.v1.beam_urn) =
+ "beam:schematransform:org.apache.beam:my_system_write:v1"];
+ }
+}
+```
+
+3. Re-generate and compile the model protos:
+```bash
+./gradlew :model:pipeline:generateProto :model:pipeline:compileJava
+```
+
+---
+
+## 4. Implementing `SchemaTransformProvider`
+
+To expose your connector to cross-language pipelines and Beam YAML, create a typed `SchemaTransformProvider`.
+
+```java
+@AutoService(SchemaTransformProvider.class)
+public class MyReadSchemaTransformProvider extends TypedSchemaTransformProvider<Configuration> {
+
+ @Override
+ public String identifier() {
+ return getUrn(ExternalTransforms.ManagedTransforms.Urns.MY_SYSTEM_READ);
+ }
+
+ @Override
+ public String description() {
+ return "Reads records from My System and outputs a PCollection of Beam Rows.";
+ }
+
+ @Override
+ protected SchemaTransform from(Configuration configuration) {
+ return new MyReadSchemaTransform(configuration);
+ }
+
+ @Override
+ public List<String> outputCollectionNames() {
+ return Collections.singletonList("output");
+ }
+
+ @DefaultSchema(AutoValueSchema.class)
+ @AutoValue
+ public abstract static class Configuration {
+ @SchemaFieldDescription("Configuration option description.")
+ public abstract String getConfigurationOption();
+
+ public static Builder builder() {
+ return new AutoValue_MyReadSchemaTransformProvider_Configuration.Builder();
+ }
+
+ @AutoValue.Builder
+ public abstract static class Builder {
+ public abstract Builder setConfigurationOption(String value);
+ public abstract Configuration build();
+ }
+ }
+
+ static class MyReadSchemaTransform extends SchemaTransform {
+ private final Configuration configuration;
+
+ MyReadSchemaTransform(Configuration configuration) {
+ this.configuration = Objects.requireNonNull(configuration, "configuration cannot be null");
+ }
+
+ @Override
+ public PCollectionRowTuple expand(PCollectionRowTuple input) {
+ PCollection<Row> output = input.getPipeline().apply(
+ MyIO.readRows().withConfigurationOption(configuration.getConfigurationOption()));
+ return PCollectionRowTuple.of("output", output);
+ }
+ }
+}
+```
+
+---
+
+## 5. Integrating with Managed API (`Managed.java`)
+
+Beam's `Managed` I/O transform provides a unified interface for data ingest/egress. To support it:
+
+1. Open `sdks/java/managed/src/main/java/org/apache/beam/sdk/managed/Managed.java`.
+2. Define a public constant identifier:
+```java
+public static final String MY_SYSTEM = "my_system";
+```
+3. Register your URNs in `READ_TRANSFORMS` or `WRITE_TRANSFORMS`:
+```java
+public static final Map<String, String> READ_TRANSFORMS =
+ ImmutableMap.<String, String>builder()
+ // ... existing transforms
+ .put(MY_SYSTEM, getUrn(ExternalTransforms.ManagedTransforms.Urns.MY_SYSTEM_READ))
+ .build();
+```
+4. Update the Javadoc block in `Managed.java` to list your new connector.
+
+---
+
+## 6. Expansion Service Registration
+
+To enable non-Java SDKs (Python, Go) to discover and expand your new connector, include it in the standard Java Expansion Service.
+
+1. Open `sdks/java/io/expansion-service/build.gradle`.
+2. Add your module as a runtime dependency:
+```groovy
+dependencies {
+ // ... existing dependencies
+ runtimeOnly project(":sdks:java:io:<connector-name>")
+}
+```
+
+---
+
+## 7. Python & Beam YAML Integration
+
+Once registered in the expansion service, your `SchemaTransform` can be utilized in Python and YAML. E.g., for `Managed` support in Python:
+
+1. Open `sdks/python/apache_beam/transforms/managed.py`.
+2. Export your identifier in `__all__` and map it to its URN in `Read._READ_TRANSFORMS` or `Write._WRITE_TRANSFORMS`:
+```python
+MY_SYSTEM = 'my_system'
+
+__all__ = [
+ # ... existing
+ "MY_SYSTEM",
+]
+
+class Read(PTransform):
+ _READ_TRANSFORMS = {
+ # ... existing
+ MY_SYSTEM: ManagedTransforms.Urns.MY_SYSTEM_READ.urn,
+ }
+```
+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`.
+
+---
+
+## 8. Verification and Testing
+
+Verify your new connector thoroughly across multiple abstraction layers:
+
+### 1. Unit Tests
+Test your core builder and `SchemaTransformProvider` translation.
+```bash
+./gradlew :sdks:java:io:<connector-name>:test
+```
+
+### 2. Managed API Translation
+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.
+```bash
+./gradlew :sdks:java:managed:test
+```
+
+### 3. Integration Tests (IT)
+Create integration tests to test end-to-end data processing against real system instances, including `Managed.read(Managed.MY_SYSTEM)` usage.
+
+* Test resources should be managed via `ResourceManager` classes under the `it/` directory.
+* 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.
+
+### 4. Documentation
+Add any necessary documentation for your connector under the `http://website/www/site/content/en/documentation/io/built-in/` directory.
+
+---
+
+> [!TIP]
+> **Canonical Reference Implementations:** When developing a new connector, we highly recommend studying **Apache Iceberg** ([IcebergIO.java](file:///Users/chamikara/code/beam_checkout_3/beam/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/IcebergIO.java)) and **Delta Lake** ([DeltaIO.java](file:///Users/chamikara/code/beam_checkout_3/beam/sdks/java/io/delta/src/main/java/org/apache/beam/sdk/io/delta/DeltaIO.java)) as state-of-the-art reference implementations.
Review Comment:
The links to `IcebergIO.java` and `DeltaIO.java` use absolute local `file:///` URLs pointing to a specific user's local directory. These should be updated to public GitHub URLs so they work for all users.
```suggestion
> **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.
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Adds a new agent SKILL for developing new I/O connectors [beam]
Posted by "chamikaramj (via GitHub)" <gi...@apache.org>.
chamikaramj commented on code in PR #38910:
URL: https://github.com/apache/beam/pull/38910#discussion_r3392245581
##########
.agent/skills/developing-new-io-connectors/SKILL.md:
##########
@@ -0,0 +1,345 @@
+---
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+name: developing-new-io-connectors
+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.
+---
+
+# Developing New Apache Beam I/O Connectors
+
+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.
+
+## 1. Module Structure and Gradle Setup
+
+New Java I/O connectors should reside under `sdks/java/io/<connector-name>`.
+
+### Directory Layout
+```text
+sdks/java/io/<connector-name>/
+├── build.gradle
+└── src/
+ ├── main/java/org/apache/beam/sdk/io/<connector-name>/
+ │ ├── <Connector>IO.java
+ │ ├── <Connector>ReadSchemaTransformProvider.java
+ │ └── <Connector>WriteSchemaTransformProvider.java
+ └── test/java/org/apache/beam/sdk/io/<connector-name>/
+ ├── <Connector>IOTest.java
+ └── <Connector>ReadSchemaTransformProviderTest.java
+```
+
+### Gradle Configuration (`build.gradle`)
+Your `build.gradle` must use standard Beam Java module conventions and explicitly declare necessary dependencies.
+
+```groovy
+plugins { id 'org.apache.beam.module' }
+applyJavaNature(
+ // If <connector-name> contains hyphens, convert them to dots or underscores
+ automaticModuleName: 'org.apache.beam.sdk.io.<connector_name>',
+)
+
+description = "Apache Beam :: SDKs :: Java :: IO :: <Connector Name>"
+ext.summary = "Integration with <External System>."
+
+dependencies {
+ implementation project(path: ":sdks:java:core", configuration: "shadow")
+ implementation project(path: ":model:pipeline", configuration: "shadow") // For URN definitions
+
+ // Add external client libraries here
+ implementation library.java.<client_dependency>
+
+ // Handle strict dependency checking if necessary
+ permitUnusedDeclared library.java.<client_dependency>
+
+ // Standard test dependencies
+ testImplementation project(path: ":sdks:java:core", configuration: "shadowTest")
+ testImplementation library.java.junit
+}
+```
+
+---
+
+## 2. Core I/O Transform Implementation (`<Connector>IO.java`)
+
+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`.
+
+### Bounded & Unbounded Sources
+Instead of legacy `Source` classes, implement reading via Beam's **Splittable DoFns (SDF)** framework for advanced features such as dynamic rebalancing and watermark support.
+
+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).
+
+An example SDF-based read transform is given below:
+
+```java
+public class MyIO {
+ public static ReadRows readRows() {
+ return new AutoValue_MyIO_ReadRows.Builder().build();
+ }
+
+ @AutoValue
+ public abstract static class ReadRows extends PTransform<PBegin, PCollection<Row>> {
+ public abstract @Nullable String getConfigurationOption();
+ public abstract @Nullable Schema getSchema();
+ public abstract Builder toBuilder();
+
+ @AutoValue.Builder
+ public abstract static class Builder {
+ public abstract Builder setConfigurationOption(String value);
+ public abstract Builder setSchema(Schema schema);
+ public abstract ReadRows build();
+ }
+
+ public ReadRows withConfigurationOption(String value) {
+ return toBuilder().setConfigurationOption(value).build();
+ }
+
+ public ReadRows withSchema(Schema schema) {
+ return toBuilder().setSchema(schema).build();
+ }
+
+ @Override
+ public PCollection<Row> expand(PBegin input) {
+ return input
+ // `ReaderDoFn` is an SDF or source implementation that reads records and outputs `Row` objects.
+ .apply(ParDo.of(new ReaderDoFn(getConfigurationOption())))
+ .setRowSchema(getSchema());
+ }
+ }
+
+ public static WriteRows writeRows() {
+ return new AutoValue_MyIO_WriteRows.Builder().build();
+ }
+
+ @AutoValue
+ public abstract static class WriteRows extends PTransform<PCollection<Row>, PDone> {
+ public abstract @Nullable String getConfigurationOption();
+ public abstract Builder toBuilder();
+
+ @AutoValue.Builder
+ public abstract static class Builder {
+ public abstract Builder setConfigurationOption(String value);
+ public abstract WriteRows build();
+ }
+
+ public WriteRows withConfigurationOption(String value) {
+ return toBuilder().setConfigurationOption(value).build();
+ }
+
+ @Override
+ public PDone expand(PCollection<Row> input) {
+ input.apply("WriteRecords", ParDo.of(new WriterDoFn(getConfigurationOption())));
+ return PDone.in(input.getPipeline());
+ }
+ }
+}
+```
+* 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/
+
+* 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.
+
+### Developing Sinks (Write Transforms)
+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:
+
+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.
+2. **Two-Phase Commit / Exactly-Once Sinks:** For transactional sinks (e.g., Relational DBs, Apache Iceberg, Delta Lake), implement a multi-stage `PTransform`:
+ * **Write Shards:** Write records in parallel tasks to staging files or temporary transactions, emitting commit descriptors (`PCollection<CommitMessage>`).
+ * **Global Commit:** Group the commit messages and execute a single final transaction to commit all staged files/shards.
+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.
+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.
+
+* 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.
+---
+
+## 3. Extending the Pipeline Model Proto (`external_transforms.proto`)
+
+To standardize your transform identifier across SDKs, define its URN in Beam's protobuf schema.
+
+1. Open `model/pipeline/src/main/proto/org/apache/beam/model/pipeline/v1/external_transforms.proto`.
+2. Add your read/write URNs under the appropriate enum (e.g., `ManagedTransforms.Urns`).
+
+```protobuf
+message ManagedTransforms {
+ enum Urns {
+ // ... existing entries
+ MY_SYSTEM_READ = 15 [(org.apache.beam.model.pipeline.v1.beam_urn) =
+ "beam:schematransform:org.apache.beam:my_system_read:v1"];
+ MY_SYSTEM_WRITE = 16 [(org.apache.beam.model.pipeline.v1.beam_urn) =
+ "beam:schematransform:org.apache.beam:my_system_write:v1"];
+ }
+}
+```
+
+3. Re-generate and compile the model protos:
+```bash
+./gradlew :model:pipeline:generateProto :model:pipeline:compileJava
+```
+
+---
+
+## 4. Implementing `SchemaTransformProvider`
+
+To expose your connector to cross-language pipelines and Beam YAML, create a typed `SchemaTransformProvider`.
+
+```java
+@AutoService(SchemaTransformProvider.class)
+public class MyReadSchemaTransformProvider extends TypedSchemaTransformProvider<Configuration> {
+
+ @Override
+ public String identifier() {
+ return getUrn(ExternalTransforms.ManagedTransforms.Urns.MY_SYSTEM_READ);
+ }
+
+ @Override
+ public String description() {
+ return "Reads records from My System and outputs a PCollection of Beam Rows.";
+ }
+
+ @Override
+ protected SchemaTransform from(Configuration configuration) {
+ return new MyReadSchemaTransform(configuration);
+ }
+
+ @Override
+ public List<String> outputCollectionNames() {
+ return Collections.singletonList("output");
+ }
+
+ @DefaultSchema(AutoValueSchema.class)
+ @AutoValue
+ public abstract static class Configuration {
+ @SchemaFieldDescription("Configuration option description.")
+ public abstract String getConfigurationOption();
+
+ public static Builder builder() {
+ return new AutoValue_MyReadSchemaTransformProvider_Configuration.Builder();
+ }
+
+ @AutoValue.Builder
+ public abstract static class Builder {
+ public abstract Builder setConfigurationOption(String value);
+ public abstract Configuration build();
+ }
+ }
+
+ static class MyReadSchemaTransform extends SchemaTransform {
+ private final Configuration configuration;
+
+ MyReadSchemaTransform(Configuration configuration) {
+ this.configuration = Objects.requireNonNull(configuration, "configuration cannot be null");
+ }
+
+ @Override
+ public PCollectionRowTuple expand(PCollectionRowTuple input) {
+ PCollection<Row> output = input.getPipeline().apply(
+ MyIO.readRows().withConfigurationOption(configuration.getConfigurationOption()));
+ return PCollectionRowTuple.of("output", output);
+ }
+ }
+}
+```
+
+---
+
+## 5. Integrating with Managed API (`Managed.java`)
+
+Beam's `Managed` I/O transform provides a unified interface for data ingest/egress. To support it:
+
+1. Open `sdks/java/managed/src/main/java/org/apache/beam/sdk/managed/Managed.java`.
+2. Define a public constant identifier:
+```java
+public static final String MY_SYSTEM = "my_system";
+```
+3. Register your URNs in `READ_TRANSFORMS` or `WRITE_TRANSFORMS`:
+```java
+public static final Map<String, String> READ_TRANSFORMS =
+ ImmutableMap.<String, String>builder()
+ // ... existing transforms
+ .put(MY_SYSTEM, getUrn(ExternalTransforms.ManagedTransforms.Urns.MY_SYSTEM_READ))
+ .build();
+```
+4. Update the Javadoc block in `Managed.java` to list your new connector.
+
+---
+
+## 6. Expansion Service Registration
+
+To enable non-Java SDKs (Python, Go) to discover and expand your new connector, include it in the standard Java Expansion Service.
+
+1. Open `sdks/java/io/expansion-service/build.gradle`.
+2. Add your module as a runtime dependency:
+```groovy
+dependencies {
+ // ... existing dependencies
+ runtimeOnly project(":sdks:java:io:<connector-name>")
+}
+```
+
+---
+
+## 7. Python & Beam YAML Integration
+
+Once registered in the expansion service, your `SchemaTransform` can be utilized in Python and YAML. E.g., for `Managed` support in Python:
+
+1. Open `sdks/python/apache_beam/transforms/managed.py`.
+2. Export your identifier in `__all__` and map it to its URN in `Read._READ_TRANSFORMS` or `Write._WRITE_TRANSFORMS`:
+```python
+MY_SYSTEM = 'my_system'
+
+__all__ = [
+ # ... existing
+ "MY_SYSTEM",
+]
+
+class Read(PTransform):
+ _READ_TRANSFORMS = {
+ # ... existing
+ MY_SYSTEM: ManagedTransforms.Urns.MY_SYSTEM_READ.urn,
+ }
+```
+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`.
+
+---
+
+## 8. Verification and Testing
+
+Verify your new connector thoroughly across multiple abstraction layers:
+
+### 1. Unit Tests
+Test your core builder and `SchemaTransformProvider` translation.
+```bash
+./gradlew :sdks:java:io:<connector-name>:test
+```
+
+### 2. Managed API Translation
+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.
+```bash
+./gradlew :sdks:java:managed:test
+```
+
+### 3. Integration Tests (IT)
+Create integration tests to test end-to-end data processing against real system instances, including `Managed.read(Managed.MY_SYSTEM)` usage.
+
+* Test resources should be managed via `ResourceManager` classes under the `it/` directory.
+* 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.
+
+### 4. Documentation
+Add any necessary documentation for your connector under the `http://website/www/site/content/en/documentation/io/built-in/` directory.
Review Comment:
Done.
##########
.agent/skills/developing-new-io-connectors/SKILL.md:
##########
@@ -0,0 +1,345 @@
+---
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+name: developing-new-io-connectors
+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.
+---
+
+# Developing New Apache Beam I/O Connectors
+
+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.
+
+## 1. Module Structure and Gradle Setup
+
+New Java I/O connectors should reside under `sdks/java/io/<connector-name>`.
+
+### Directory Layout
+```text
+sdks/java/io/<connector-name>/
+├── build.gradle
+└── src/
+ ├── main/java/org/apache/beam/sdk/io/<connector-name>/
+ │ ├── <Connector>IO.java
+ │ ├── <Connector>ReadSchemaTransformProvider.java
+ │ └── <Connector>WriteSchemaTransformProvider.java
+ └── test/java/org/apache/beam/sdk/io/<connector-name>/
+ ├── <Connector>IOTest.java
+ └── <Connector>ReadSchemaTransformProviderTest.java
+```
+
+### Gradle Configuration (`build.gradle`)
+Your `build.gradle` must use standard Beam Java module conventions and explicitly declare necessary dependencies.
+
+```groovy
+plugins { id 'org.apache.beam.module' }
+applyJavaNature(
+ // If <connector-name> contains hyphens, convert them to dots or underscores
+ automaticModuleName: 'org.apache.beam.sdk.io.<connector_name>',
+)
+
+description = "Apache Beam :: SDKs :: Java :: IO :: <Connector Name>"
+ext.summary = "Integration with <External System>."
+
+dependencies {
+ implementation project(path: ":sdks:java:core", configuration: "shadow")
+ implementation project(path: ":model:pipeline", configuration: "shadow") // For URN definitions
+
+ // Add external client libraries here
+ implementation library.java.<client_dependency>
+
+ // Handle strict dependency checking if necessary
+ permitUnusedDeclared library.java.<client_dependency>
+
+ // Standard test dependencies
+ testImplementation project(path: ":sdks:java:core", configuration: "shadowTest")
+ testImplementation library.java.junit
+}
+```
+
+---
+
+## 2. Core I/O Transform Implementation (`<Connector>IO.java`)
+
+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`.
+
+### Bounded & Unbounded Sources
+Instead of legacy `Source` classes, implement reading via Beam's **Splittable DoFns (SDF)** framework for advanced features such as dynamic rebalancing and watermark support.
+
+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).
+
+An example SDF-based read transform is given below:
+
+```java
+public class MyIO {
+ public static ReadRows readRows() {
+ return new AutoValue_MyIO_ReadRows.Builder().build();
+ }
+
+ @AutoValue
+ public abstract static class ReadRows extends PTransform<PBegin, PCollection<Row>> {
+ public abstract @Nullable String getConfigurationOption();
+ public abstract @Nullable Schema getSchema();
+ public abstract Builder toBuilder();
+
+ @AutoValue.Builder
+ public abstract static class Builder {
+ public abstract Builder setConfigurationOption(String value);
+ public abstract Builder setSchema(Schema schema);
+ public abstract ReadRows build();
+ }
+
+ public ReadRows withConfigurationOption(String value) {
+ return toBuilder().setConfigurationOption(value).build();
+ }
+
+ public ReadRows withSchema(Schema schema) {
+ return toBuilder().setSchema(schema).build();
+ }
+
+ @Override
+ public PCollection<Row> expand(PBegin input) {
+ return input
+ // `ReaderDoFn` is an SDF or source implementation that reads records and outputs `Row` objects.
+ .apply(ParDo.of(new ReaderDoFn(getConfigurationOption())))
+ .setRowSchema(getSchema());
+ }
+ }
+
+ public static WriteRows writeRows() {
+ return new AutoValue_MyIO_WriteRows.Builder().build();
+ }
+
+ @AutoValue
+ public abstract static class WriteRows extends PTransform<PCollection<Row>, PDone> {
+ public abstract @Nullable String getConfigurationOption();
+ public abstract Builder toBuilder();
+
+ @AutoValue.Builder
+ public abstract static class Builder {
+ public abstract Builder setConfigurationOption(String value);
+ public abstract WriteRows build();
+ }
+
+ public WriteRows withConfigurationOption(String value) {
+ return toBuilder().setConfigurationOption(value).build();
+ }
+
+ @Override
+ public PDone expand(PCollection<Row> input) {
+ input.apply("WriteRecords", ParDo.of(new WriterDoFn(getConfigurationOption())));
+ return PDone.in(input.getPipeline());
+ }
+ }
+}
+```
+* 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/
+
+* 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.
+
+### Developing Sinks (Write Transforms)
+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:
+
+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.
+2. **Two-Phase Commit / Exactly-Once Sinks:** For transactional sinks (e.g., Relational DBs, Apache Iceberg, Delta Lake), implement a multi-stage `PTransform`:
+ * **Write Shards:** Write records in parallel tasks to staging files or temporary transactions, emitting commit descriptors (`PCollection<CommitMessage>`).
+ * **Global Commit:** Group the commit messages and execute a single final transaction to commit all staged files/shards.
+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.
+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.
+
+* 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.
+---
+
+## 3. Extending the Pipeline Model Proto (`external_transforms.proto`)
+
+To standardize your transform identifier across SDKs, define its URN in Beam's protobuf schema.
+
+1. Open `model/pipeline/src/main/proto/org/apache/beam/model/pipeline/v1/external_transforms.proto`.
+2. Add your read/write URNs under the appropriate enum (e.g., `ManagedTransforms.Urns`).
+
+```protobuf
+message ManagedTransforms {
+ enum Urns {
+ // ... existing entries
+ MY_SYSTEM_READ = 15 [(org.apache.beam.model.pipeline.v1.beam_urn) =
+ "beam:schematransform:org.apache.beam:my_system_read:v1"];
+ MY_SYSTEM_WRITE = 16 [(org.apache.beam.model.pipeline.v1.beam_urn) =
+ "beam:schematransform:org.apache.beam:my_system_write:v1"];
+ }
+}
+```
+
+3. Re-generate and compile the model protos:
+```bash
+./gradlew :model:pipeline:generateProto :model:pipeline:compileJava
+```
+
+---
+
+## 4. Implementing `SchemaTransformProvider`
+
+To expose your connector to cross-language pipelines and Beam YAML, create a typed `SchemaTransformProvider`.
+
+```java
+@AutoService(SchemaTransformProvider.class)
+public class MyReadSchemaTransformProvider extends TypedSchemaTransformProvider<Configuration> {
+
+ @Override
+ public String identifier() {
+ return getUrn(ExternalTransforms.ManagedTransforms.Urns.MY_SYSTEM_READ);
+ }
+
+ @Override
+ public String description() {
+ return "Reads records from My System and outputs a PCollection of Beam Rows.";
+ }
+
+ @Override
+ protected SchemaTransform from(Configuration configuration) {
+ return new MyReadSchemaTransform(configuration);
+ }
+
+ @Override
+ public List<String> outputCollectionNames() {
+ return Collections.singletonList("output");
+ }
+
+ @DefaultSchema(AutoValueSchema.class)
+ @AutoValue
+ public abstract static class Configuration {
+ @SchemaFieldDescription("Configuration option description.")
+ public abstract String getConfigurationOption();
+
+ public static Builder builder() {
+ return new AutoValue_MyReadSchemaTransformProvider_Configuration.Builder();
+ }
+
+ @AutoValue.Builder
+ public abstract static class Builder {
+ public abstract Builder setConfigurationOption(String value);
+ public abstract Configuration build();
+ }
+ }
+
+ static class MyReadSchemaTransform extends SchemaTransform {
+ private final Configuration configuration;
+
+ MyReadSchemaTransform(Configuration configuration) {
+ this.configuration = Objects.requireNonNull(configuration, "configuration cannot be null");
+ }
+
+ @Override
+ public PCollectionRowTuple expand(PCollectionRowTuple input) {
+ PCollection<Row> output = input.getPipeline().apply(
+ MyIO.readRows().withConfigurationOption(configuration.getConfigurationOption()));
+ return PCollectionRowTuple.of("output", output);
+ }
+ }
+}
+```
+
+---
+
+## 5. Integrating with Managed API (`Managed.java`)
+
+Beam's `Managed` I/O transform provides a unified interface for data ingest/egress. To support it:
+
+1. Open `sdks/java/managed/src/main/java/org/apache/beam/sdk/managed/Managed.java`.
+2. Define a public constant identifier:
+```java
+public static final String MY_SYSTEM = "my_system";
+```
+3. Register your URNs in `READ_TRANSFORMS` or `WRITE_TRANSFORMS`:
+```java
+public static final Map<String, String> READ_TRANSFORMS =
+ ImmutableMap.<String, String>builder()
+ // ... existing transforms
+ .put(MY_SYSTEM, getUrn(ExternalTransforms.ManagedTransforms.Urns.MY_SYSTEM_READ))
+ .build();
+```
+4. Update the Javadoc block in `Managed.java` to list your new connector.
+
+---
+
+## 6. Expansion Service Registration
+
+To enable non-Java SDKs (Python, Go) to discover and expand your new connector, include it in the standard Java Expansion Service.
+
+1. Open `sdks/java/io/expansion-service/build.gradle`.
+2. Add your module as a runtime dependency:
+```groovy
+dependencies {
+ // ... existing dependencies
+ runtimeOnly project(":sdks:java:io:<connector-name>")
+}
+```
+
+---
+
+## 7. Python & Beam YAML Integration
+
+Once registered in the expansion service, your `SchemaTransform` can be utilized in Python and YAML. E.g., for `Managed` support in Python:
+
+1. Open `sdks/python/apache_beam/transforms/managed.py`.
+2. Export your identifier in `__all__` and map it to its URN in `Read._READ_TRANSFORMS` or `Write._WRITE_TRANSFORMS`:
+```python
+MY_SYSTEM = 'my_system'
+
+__all__ = [
+ # ... existing
+ "MY_SYSTEM",
+]
+
+class Read(PTransform):
+ _READ_TRANSFORMS = {
+ # ... existing
+ MY_SYSTEM: ManagedTransforms.Urns.MY_SYSTEM_READ.urn,
+ }
+```
+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`.
+
+---
+
+## 8. Verification and Testing
+
+Verify your new connector thoroughly across multiple abstraction layers:
+
+### 1. Unit Tests
+Test your core builder and `SchemaTransformProvider` translation.
+```bash
+./gradlew :sdks:java:io:<connector-name>:test
+```
+
+### 2. Managed API Translation
+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.
+```bash
+./gradlew :sdks:java:managed:test
+```
+
+### 3. Integration Tests (IT)
+Create integration tests to test end-to-end data processing against real system instances, including `Managed.read(Managed.MY_SYSTEM)` usage.
+
+* Test resources should be managed via `ResourceManager` classes under the `it/` directory.
+* 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.
+
+### 4. Documentation
+Add any necessary documentation for your connector under the `http://website/www/site/content/en/documentation/io/built-in/` directory.
+
+---
+
+> [!TIP]
+> **Canonical Reference Implementations:** When developing a new connector, we highly recommend studying **Apache Iceberg** ([IcebergIO.java](file:///Users/chamikara/code/beam_checkout_3/beam/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/IcebergIO.java)) and **Delta Lake** ([DeltaIO.java](file:///Users/chamikara/code/beam_checkout_3/beam/sdks/java/io/delta/src/main/java/org/apache/beam/sdk/io/delta/DeltaIO.java)) as state-of-the-art reference implementations.
Review Comment:
Done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org