(beam) branch master updated: Add Iceberg to AlloyDB Blueprint (#36036)

[xq...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

xqhu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git


The following commit(s) were added to refs/heads/master by this push:
     new c367bd3347e Add Iceberg to AlloyDB Blueprint (#36036)
c367bd3347e is described below

commit c367bd3347e41473c15ef8968ca43283221cb2e7
Author: Tarun Annapareddy <ta...@google.com>
AuthorDate: Thu Sep 4 10:22:29 2025 -0700

    Add Iceberg to AlloyDB Blueprint (#36036)
    
    * Add Iceberg to AlloyDB Blueprint
    
    * use WriteToJdbc
    
    * Fix Formatting
    
    * Fix description
    
    * Add place holders for creds
---
 .../yaml/examples/testing/examples_test.py         |  6 ++-
 .../transforms/blueprint/iceberg_to_alloydb.yaml   | 51 ++++++++++++++++++++++
 2 files changed, 55 insertions(+), 2 deletions(-)

diff --git a/sdks/python/apache_beam/yaml/examples/testing/examples_test.py b/sdks/python/apache_beam/yaml/examples/testing/examples_test.py
index 80e82945523..4f0516a1ea9 100644
--- a/sdks/python/apache_beam/yaml/examples/testing/examples_test.py
+++ b/sdks/python/apache_beam/yaml/examples/testing/examples_test.py
@@ -678,7 +678,8 @@ def _kafka_test_preprocessor(
     'test_ml_preprocessing_yaml',
     'test_anomaly_scoring_yaml',
     'test_wordCountInclude_yaml',
-    'test_wordCountImport_yaml'
+    'test_wordCountImport_yaml',
+    'test_iceberg_to_alloydb_yaml'
 ])
 def _io_write_test_preprocessor(
     test_spec: dict, expected: List[str], env: TestEnvironment):
@@ -749,7 +750,8 @@ def _file_io_read_test_preprocessor(
   return test_spec
 
 
-@YamlExamplesTestSuite.register_test_preprocessor(['test_iceberg_read_yaml'])
+@YamlExamplesTestSuite.register_test_preprocessor(
+    ['test_iceberg_read_yaml', 'test_iceberg_to_alloydb_yaml'])
 def _iceberg_io_read_test_preprocessor(
     test_spec: dict, expected: List[str], env: TestEnvironment):
   """
diff --git a/sdks/python/apache_beam/yaml/examples/transforms/blueprint/iceberg_to_alloydb.yaml b/sdks/python/apache_beam/yaml/examples/transforms/blueprint/iceberg_to_alloydb.yaml
new file mode 100644
index 00000000000..dc4a887fea1
--- /dev/null
+++ b/sdks/python/apache_beam/yaml/examples/transforms/blueprint/iceberg_to_alloydb.yaml
@@ -0,0 +1,51 @@
+# coding=utf-8
+#
+# 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.
+
+# A pipeline that reads from iceberg table and writes to AlloyDB.
+
+pipeline:
+  type: chain
+  transforms:
+    # Step 1: Reading data from Iceberg
+    - type: ReadFromIceberg
+      name: ReadFromIcebergTable
+      config:
+        table: "db.users.NY"
+        catalog_name: "hadoop_catalog"
+        catalog_properties:
+          type: "hadoop"
+          warehouse: "gs://MY-WAREHOUSE"
+        # Hadoop catalog config required to run pipeline locally
+        # Omit if running on Dataflow
+        config_properties:
+          "fs.gs.auth.type": "SERVICE_ACCOUNT_JSON_KEYFILE"
+          "fs.gs.auth.service.account.json.keyfile": "/path/to/service/account/key.json"
+
+    # Step 2: Write records out to AlloyDB
+    # To run pipeline locally, Modify alloydbIpType to PUBLIC
+    - type: WriteToJdbc
+      name: WriteToAlloyDBTable
+      config:
+        location: "users"
+        driver_class_name: "org.postgresql.Driver"
+        jdbc_url: "jdbc:postgresql:///db?socketFactory=com.google.cloud.alloydb.SocketFactory&alloydbInstanceName=projects/<PROJECT>/locations/<REGION>/clusters/<CLUSTER>/instances/<INSTANCE>&alloydbIpType=PRIVATE"
+        username: "<YOUR_USERNAME>"
+        password: "<YOUR_PASSWORD>"
+
+# Expected:
+#  Row(id=3, name='Smith', email='smith@example.com', zip='NY')
+#  Row(id=4, name='Beamberg', email='beamberg@example.com', zip='NY')