debezium/dbz#2863 add apache-beam example

.github/workflows/apache-beam-workflow.yml

@@ -0,0 +1,31 @@
+name: Build [apache-beam]
+
+on:
+  push:
+    paths:
+      - 'apache-beam/**'
+      - '.github/workflows/apache-beam-workflow.yml'
+  pull_request:
+    paths:
+      - 'apache-beam/**'
+      - '.github/workflows/apache-beam-workflow.yml'
+      - '.env'
+
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v6
+      - uses: actions/setup-java@v4
+        with:
+          distribution: 'temurin'
+          java-version: '21'
+      - name: Cache local Maven repository
+        uses: actions/cache@v4
+        with:
+          path: ~/.m2/repository
+          key: ${{ runner.os }}-maven-${{ hashFiles('apache-beam/**/pom.xml', 'pom.xml') }}
+          restore-keys: |
+            ${{ runner.os }}-maven-
+      - name: Check changes in [apache-beam] example
+        run: cd apache-beam && mvn clean install -B -Dorg.slf4j.simpleLogger.log.org.apache.maven.cli.transfer.Slf4jMavenTransferListener=warn -Dmaven.wagon.http.pool=false -Dmaven.wagon.httpconnectionManager.ttlSeconds=120

README.md

@@ -11,6 +11,7 @@ For getting started please check the [tutorial example](./tutorial).
 ## Examples
 
 * [AI RAG](./ai-rag): How to implement Retrieval-Augmented Generation (RAG) using **Debezium Server** to stream **Postgres** changes, compute embeddings, and store them in a **Milvus** vector database for use by a local **Ollama**-served LLM
+* [Apache Beam](./apache-beam): Streaming **MySQL** database changes to **Google BigQuery** with `DebeziumIO` in an **Apache Beam** pipeline
 * [Apache Pulsar](./apache-pulsar): Streaming **Postgres** database changes to **Apache Pulsar**
 * [Audit Logs](./auditlog): Building Audit Logs with Change Data Capture
 * [Cache Invalidation](./cache-invalidation): How Debezium can be used to invalidate items in the JPA 2nd level cache after external data changes

apache-beam/README.md

@@ -0,0 +1,84 @@
+# Debezium in Apache Beam
+
+An Apache Beam pipeline that uses `DebeziumIO` to stream changes from MySQL table into BigQuery.
+
+## Prerequisites
+
+* Docker and Docker Compose
+* Java 21 and Maven
+
+## Run
+
+Start MySQL and the BigQuery emulator:
+
+```shell
+docker compose --env-file ../.env up -d
+```
+
+Start the pipeline with the Direct runner:
+
+```shell
+mvn compile exec:java -Dexec.args="\
+  --runner=DirectRunner \
+  --bigQueryEndpoint=http://localhost:9050 \
+  --credentialFactoryClass=org.apache.beam.sdk.extensions.gcp.auth.NoopCredentialFactory"
+```
+
+Open http://localhost:8000 to see the rows, including the update.
+
+To get the same rows as in MySQL, keep only the latest change per customer:
+
+```sql
+WITH BASE AS (
+    SELECT
+        PARSE_JSON(`after`).fields a,
+        PARSE_JSON(`before`).fields b,
+        write_time
+    FROM my-dataset.customers
+)
+
+SELECT
+    LAX_INT64(a.id) id,
+    CASE WHEN STRING(a.first_name) = "Sally" then "Sally Anne" ELSE STRING(a.first_name) END first_name,
+    STRING(a.last_name) last_name,
+    STRING(a.email) email,
+    write_time
+FROM BASE
+         QUALIFY ROW_NUMBER() OVER(PARTITION BY LAX_INT64(coalesce(a.id, b.id)) ORDER BY write_time DESC) = 1
+  AND a IS NOT NULL
+ORDER BY 1 ASC
+```
+
+## Clean up
+
+Stop the pipeline with `Ctrl+C`, then:
+
+```shell
+docker compose --env-file ../.env down
+```
+
+## Others
+
+Start the pipeline with the Dataflow Runner:
+
+In [`Main.java`](src/main/java/io/debezium/examples/apache/beam/Main.java), replace `localhost` with a MySQL address the Dataflow workers can reach.
+
+```shell
+gcloud auth application-default login
+
+mvn compile exec:java -Dexec.args="\
+    --runner=DataflowRunner \
+    --project=my-project \
+    --region=asia-southeast2 \
+    --streaming \
+    --experiments=enable_streaming_engine \
+    --jobName=debezium-mysql-to-bq \
+    --tempLocation=gs://my-bucket/temp \
+    --stagingLocation=gs://my-bucket/staging \
+    --serviceAccount=my-service-account@my-project.iam.gserviceaccount.com \
+    --defaultSdkHarnessLogLevel=WARN \
+    --workerMachineType=n1-standard-1 \
+    --diskSizeGb=10 \
+    --numWorkers=1 \
+    --maxNumWorkers=1"
+```

apache-beam/bigquery-ui/Dockerfile

@@ -0,0 +1,6 @@
+FROM python:3.13-slim
+WORKDIR /app
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+COPY app.py .
+CMD ["python", "app.py"]

apache-beam/bigquery-ui/app.py

@@ -0,0 +1,67 @@
+import os
+
+from flask import Flask, render_template_string, request
+from google.api_core.client_options import ClientOptions
+from google.auth.credentials import AnonymousCredentials
+from google.cloud import bigquery
+
+PROJECT = os.getenv("BIGQUERY_PROJECT", "my-project")
+ENDPOINT = os.getenv("BIGQUERY_ENDPOINT", "http://localhost:9050")
+DEFAULT_QUERY = "SELECT * FROM my-dataset.customers ORDER BY write_time DESC"
+
+client = bigquery.Client(
+    PROJECT,
+    client_options=ClientOptions(api_endpoint=ENDPOINT),
+    credentials=AnonymousCredentials(),
+)
+
+app = Flask(__name__)
+
+PAGE = """<!doctype html>
+<title>BigQuery emulator</title>
+<h1>BigQuery emulator: {{ project }}</h1>
+<p>Tables:
+{% for table in tables %}
+  <a href="?query={{ ('SELECT * FROM ' ~ table) | urlencode }}">{{ table }}</a>
+{% endfor %}
+</p>
+<form>
+  <textarea name="query" rows="4" cols="100">{{ query }}</textarea><br>
+  <button>Run</button>
+</form>
+{% if error %}
+  <pre>{{ error }}</pre>
+{% else %}
+  <p>{{ rows | length }} rows</p>
+  <table border="1">
+    <tr>{% for column in columns %}<th>{{ column }}</th>{% endfor %}</tr>
+    {% for row in rows %}
+      <tr>{% for value in row %}<td>{{ value }}</td>{% endfor %}</tr>
+    {% endfor %}
+  </table>
+{% endif %}
+"""
+
+
+@app.route("/")
+def index():
+    query = request.args.get("query", DEFAULT_QUERY)
+    tables = [
+        f"{table.dataset_id}.{table.table_id}"
+        for dataset in client.list_datasets()
+        for table in client.list_tables(dataset)
+    ]
+    columns, rows, error = [], [], None
+    try:
+        result = client.query_and_wait(query, job_retry=None)
+        columns = [field.name for field in result.schema]
+        rows = [list(row.values()) for row in result]
+    except Exception as e:
+        error = str(e)
+    return render_template_string(
+        PAGE, project=PROJECT, tables=tables, query=query, columns=columns, rows=rows, error=error
+    )
+
+
+if __name__ == "__main__":
+    app.run(host="0.0.0.0", port=8000)

apache-beam/bigquery-ui/requirements.txt

@@ -0,0 +1,2 @@
+flask==3.1.3
+google-cloud-bigquery==3.46.1

apache-beam/data.yaml

@@ -0,0 +1,16 @@
+# Same table as the "Create BigQuery table" step of the blog post
+projects:
+  - id: my-project
+    datasets:
+      - id: my-dataset
+        tables:
+          - id: customers
+            columns:
+              - name: metadata
+                type: STRING
+              - name: before
+                type: STRING
+              - name: after
+                type: STRING
+              - name: write_time
+                type: NUMERIC

apache-beam/docker-compose.yaml

@@ -0,0 +1,32 @@
+services:
+
+  mysql:
+    image: quay.io/debezium/example-mysql:${DEBEZIUM_VERSION}
+    container_name: mysql-beam
+    ports:
+      - "3306:3306"
+    environment:
+      - MYSQL_ROOT_PASSWORD=debezium
+      - MYSQL_USER=mysqluser
+      - MYSQL_PASSWORD=mysqlpw
+
+  bigquery:
+    image: ghcr.io/goccy/bigquery-emulator:latest
+    container_name: bigquery-beam
+    ports:
+      - "9050:9050"
+      - "9060:9060"
+    volumes:
+      - ./data.yaml:/data.yaml
+    command: --project=test-project --data-from-yaml=/data.yaml
+
+  bigquery-ui:
+    build: ./bigquery-ui
+    pull_policy: build
+    container_name: bigquery-ui-beam
+    ports:
+      - "8000:8000"
+    environment:
+      - BIGQUERY_ENDPOINT=http://bigquery:9050
+    depends_on:
+      - bigquery

apache-beam/pom.xml

@@ -0,0 +1,72 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <groupId>io.debezium.examples</groupId>
+        <artifactId>debezium-examples-parent</artifactId>
+        <version>1.0-SNAPSHOT</version>
+        <relativePath>../pom.xml</relativePath>
+    </parent>
+
+    <groupId>io.debezium.examples.apache.beam</groupId>
+    <artifactId>apache-beam</artifactId>
+
+    <properties>
+        <version.beam>2.76.0</version.beam>
+        <version.debezium>3.5.2.Final</version.debezium>
+    </properties>
+
+    <dependencies>
+        <!--
+            Pinned to 3.5.2.Final to match the Debezium version Apache Beam 2.76.0 is built against
+            (beam-sdks-java-io-debezium depends on debezium-core 3.5.2.Final).
+            See https://repo1.maven.org/maven2/org/apache/beam/beam-sdks-java-io-debezium/2.76.0/beam-sdks-java-io-debezium-2.76.0.pom
+        -->
+        <dependency>
+            <groupId>io.debezium</groupId>
+            <artifactId>debezium-connector-mysql</artifactId>
+            <version>${version.debezium}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.beam</groupId>
+            <artifactId>beam-sdks-java-io-debezium</artifactId>
+            <version>${version.beam}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.beam</groupId>
+            <artifactId>beam-sdks-java-io-google-cloud-platform</artifactId>
+            <version>${version.beam}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.beam</groupId>
+            <artifactId>beam-sdks-java-core</artifactId>
+            <version>${version.beam}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.beam</groupId>
+            <artifactId>beam-runners-google-cloud-dataflow-java</artifactId>
+            <version>${version.beam}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.beam</groupId>
+            <artifactId>beam-runners-direct-java</artifactId>
+            <version>${version.beam}</version>
+            <scope>runtime</scope>
+        </dependency>
+    </dependencies>
+
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.codehaus.mojo</groupId>
+                <artifactId>exec-maven-plugin</artifactId>
+                <configuration>
+                    <mainClass>io.debezium.examples.apache.beam.Main</mainClass>
+                </configuration>
+            </plugin>
+        </plugins>
+    </build>
+
+</project>