feat(yaml): Add BigTable write connector by arnavarora2004 · Pull Request #35435 · apache/beam

arnavarora2004 · 2025-06-25T20:11:13Z

@damccorm @derrickaw @fozzie15 @ahmedabu98

I added the BigTable Connector for BeamYaml

added new logic for bigtable yaml with the option to simplify the user input and make it more readable than the old logic,

all mutations work correctly

added tests to integration_test.py (I commented out some bugs in integration_test.py from new commits on the main branch, will remove if no more errors)

added a new test IT class called BigTableSimpleWriteSchemaTransformProviderIT

all logic works, please let me know if anything looks off/funny and if anything can be improved

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.

See the Contributor Guide for more tips on how to make review process smoother.

To check the build health, please visit https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md

GitHub Actions Tests Status (on master branch)

See CI.md for more information about GitHub Actions CI or the workflows README to see a list of phrases to trigger workflows.

… connected and actually look good on user end for mutations

…est debugging

…test wise

…hemaTransformProviderIT, and testing out new mutations etc

…ted new user input, all mutations work correctly, put demo code for it

sdks/python/apache_beam/yaml/integration_tests.py

...ain/java/org/apache/beam/sdk/io/gcp/bigtable/BigtableSimpleWriteSchemaTransformProvider.java

...t/java/org/apache/beam/sdk/io/gcp/bigtable/BigtableSimpleWriteSchemaTransformProviderIT.java

…am/sdk/io/gcp/bigtable/BigtableSimpleWriteSchemaTransformProvider.java Co-authored-by: Derrick Williams <myutat@gmail.com>

sdks/python/apache_beam/yaml/integration_tests.py

derrickaw · 2025-07-16T22:13:31Z

sdks/python/apache_beam/yaml/tests/bigTable.yaml

+#              # Deletes all cells in a specific column, optionally within a time range.
+#              - {key: 'row2',
+#                  type: 'DeleteFromColumn',
+#                  family_name: "cf1",
+#                  column_qualifier: "cq1",
+#                  start_timestamp_micros: 2000,
+#                  end_timestamp_micros: 5000 }
+#
+#              # Deletes all cells in a specific column family.
+#              - {key: 'row3',
+#                  type: 'DeleteFromFamily',
+#                  family_name: "cf2" }
+#
+#              # Deletes all cells in a specific row.
+#              - {key: 'row4',
+#                  type: 'DeleteFromRow' }


should we use this or create another pipeline?

derrickaw · 2025-07-16T22:13:55Z

sdks/python/apache_beam/yaml/tests/bigTable.yaml

+#  - pipeline:
+#      type: chain
+#      transforms:
+#        - type: ReadFromBigTable
+#          config:
+#            project: 'apache-beam-testing'
+#            instance: 'bt-write-tests'
+#            table: 'test-table'
+#        - type: MapToFields # Convert bytes back to strings for comparison
+#          name: ConvertBytesToStrings
+#          config:
+#            language: python
+#            fields:
+#              key:
+#                callable: |
+#                  def convert_to_string(row):
+#                    return row.key.decode('utf-8') if row.key is not None else None
+#              family_name:
+#                callable: |
+#                  def convert_to_string(row):
+#                    return row.family_name.decode('utf-8') if row.family_name is not None else None
+#              column_qualifier:
+#                callable: |
+#                  def convert_to_string(row):
+#                    return row.column_qualifier.decode('utf-8') if row.column_qualifier is not None else None
+#              value:
+#                callable: |
+#                  def convert_to_string(row):
+#                    return row.value.decode('utf-8') if row.value is not None else None
+#        - type: AssertEqual
+#          config:
+#            elements:


can we use ReadFromBigTable?

sdks/python/apache_beam/yaml/integration_tests.py

ahmedabu98 · 2025-07-17T14:48:07Z