From 33f3988b187fbf5973e26fe7f9eeb1896a78cbda Mon Sep 17 00:00:00 2001 From: Ilyas Ahsan Date: Wed, 14 Oct 2026 15:22:21 +0700 Subject: [PATCH] debezium/dbz#2863 add apache-beam example Signed-off-by: Ilyas Ahsan --- .github/workflows/apache-beam-workflow.yml | 31 +++++++ README.md | 1 + apache-beam/README.md | 84 ++++++++++++++++++ apache-beam/bigquery-ui/Dockerfile | 6 ++ apache-beam/bigquery-ui/app.py | 67 +++++++++++++++ apache-beam/bigquery-ui/requirements.txt | 2 + apache-beam/data.yaml | 16 ++++ apache-beam/docker-compose.yaml | 32 +++++++ apache-beam/pom.xml | 72 ++++++++++++++++ .../debezium/examples/apache/beam/Main.java | 86 +++++++++++++++++++ 10 files changed, 397 insertions(+) create mode 100644 .github/workflows/apache-beam-workflow.yml create mode 100644 apache-beam/README.md create mode 100644 apache-beam/bigquery-ui/Dockerfile create mode 100644 apache-beam/bigquery-ui/app.py create mode 100644 apache-beam/bigquery-ui/requirements.txt create mode 100644 apache-beam/data.yaml create mode 100644 apache-beam/docker-compose.yaml create mode 100644 apache-beam/pom.xml create mode 100644 apache-beam/src/main/java/io/debezium/examples/apache/beam/Main.java diff --git a/.github/workflows/apache-beam-workflow.yml b/.github/workflows/apache-beam-workflow.yml new file mode 100644 index 0000000000..bb459ec973 --- /dev/null +++ b/.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 diff --git a/README.md b/README.md index 52836beffb..3c3470b8ea 100644 --- a/README.md +++ b/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 diff --git a/apache-beam/README.md b/apache-beam/README.md new file mode 100644 index 0000000000..540568fce0 --- /dev/null +++ b/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" +``` diff --git a/apache-beam/bigquery-ui/Dockerfile b/apache-beam/bigquery-ui/Dockerfile new file mode 100644 index 0000000000..015e8d9673 --- /dev/null +++ b/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"] diff --git a/apache-beam/bigquery-ui/app.py b/apache-beam/bigquery-ui/app.py new file mode 100644 index 0000000000..3d7fa4f637 --- /dev/null +++ b/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 = """ +BigQuery emulator +

BigQuery emulator: {{ project }}

+

Tables: +{% for table in tables %} + {{ table }} +{% endfor %} +

+
+
+ +
+{% if error %} + 
{{ error }}
+{% else %} +

{{ rows | length }} rows

+ + {% for column in columns %}{% endfor %} + {% for row in rows %} + {% for value in row %}{% endfor %} + {% endfor %} +
{{ column }}
{{ value }}
+{% 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) diff --git a/apache-beam/bigquery-ui/requirements.txt b/apache-beam/bigquery-ui/requirements.txt new file mode 100644 index 0000000000..79a297d06e --- /dev/null +++ b/apache-beam/bigquery-ui/requirements.txt @@ -0,0 +1,2 @@ +flask==3.1.3 +google-cloud-bigquery==3.46.1 diff --git a/apache-beam/data.yaml b/apache-beam/data.yaml new file mode 100644 index 0000000000..6f8f061293 --- /dev/null +++ b/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 diff --git a/apache-beam/docker-compose.yaml b/apache-beam/docker-compose.yaml new file mode 100644 index 0000000000..c3e71f7a8b --- /dev/null +++ b/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 diff --git a/apache-beam/pom.xml b/apache-beam/pom.xml new file mode 100644 index 0000000000..4ce58f165d --- /dev/null +++ b/apache-beam/pom.xml @@ -0,0 +1,72 @@ + + + 4.0.0 + + io.debezium.examples + debezium-examples-parent + 1.0-SNAPSHOT + ../pom.xml + + + io.debezium.examples.apache.beam + apache-beam + + + 2.76.0 + 3.5.2.Final + + + + + + io.debezium + debezium-connector-mysql + ${version.debezium} + + + org.apache.beam + beam-sdks-java-io-debezium + ${version.beam} + + + org.apache.beam + beam-sdks-java-io-google-cloud-platform + ${version.beam} + + + org.apache.beam + beam-sdks-java-core + ${version.beam} + + + org.apache.beam + beam-runners-google-cloud-dataflow-java + ${version.beam} + + + org.apache.beam + beam-runners-direct-java + ${version.beam} + runtime + + + + + + + org.codehaus.mojo + exec-maven-plugin + + io.debezium.examples.apache.beam.Main + + + + + + diff --git a/apache-beam/src/main/java/io/debezium/examples/apache/beam/Main.java b/apache-beam/src/main/java/io/debezium/examples/apache/beam/Main.java new file mode 100644 index 0000000000..e77c4f5cb6 --- /dev/null +++ b/apache-beam/src/main/java/io/debezium/examples/apache/beam/Main.java @@ -0,0 +1,86 @@ +package io.debezium.examples.apache.beam; + + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.google.api.services.bigquery.model.TableReference; +import com.google.api.services.bigquery.model.TableRow; +import io.debezium.connector.mysql.MySqlConnector; +import org.apache.beam.io.debezium.DebeziumIO; +import org.apache.beam.io.debezium.DebeziumIO.ConnectorConfiguration; +import org.apache.beam.io.debezium.SourceRecordJson; +import org.apache.beam.sdk.Pipeline; +import org.apache.beam.sdk.coders.StringUtf8Coder; +import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO; +import org.apache.beam.sdk.io.gcp.bigquery.TableDestination; +import org.apache.beam.sdk.options.PipelineOptionsFactory; +import org.apache.beam.sdk.transforms.DoFn; +import org.apache.beam.sdk.transforms.ParDo; +import org.apache.beam.sdk.transforms.SerializableFunction; +import org.apache.beam.sdk.values.ValueInSingleWindow; +import org.joda.time.DateTimeUtils; + +import java.io.IOException; +import java.util.Objects; + +public class Main { + + public static void main(String[] args) { + Pipeline pipeline = Pipeline.create(PipelineOptionsFactory.fromArgs(args).withValidation().create()); + + pipeline + // Step 1. Read change events with Debezium + .apply("ReadDatabaseEvents", DebeziumIO.readAsJson() + .withConnectorConfiguration(ConnectorConfiguration.create() + .withConnectorClass(MySqlConnector.class) + .withHostName("localhost") + .withPort("3306") + .withUsername("debezium") + .withPassword("dbz") + .withConnectionProperty("database.server.id", "184054") + .withConnectionProperty("database.include.list", "inventory") + .withConnectionProperty("table.include.list", "inventory.customers") + .withConnectionProperty("include.schema.changes", "false") + .withConnectionProperty("snapshot.mode", "initial") + .withConnectionProperty("snapshot.locking.mode", "none") + .withConnectionProperty("tombstones.on.delete", "false") + ) + .withFormatFunction(new SourceRecordJson.SourceRecordJsonMapper()) + .withCoder(StringUtf8Coder.of()) + ) + + // Step 2. Transform the records into TableRow + .apply("TransformToTableRow", ParDo.of(new DoFn() { + @ProcessElement + public void processElement(ProcessContext c) throws IOException { + JsonNode payload = new ObjectMapper().readTree(c.element()); + TableRow record = new TableRow() + .set("metadata", payload.get("metadata").toString()) + .set("before", payload.get("before").toString()) + .set("after", payload.get("after").toString()) + .set("table_id", payload.get("metadata").get("table").asText()) + .set("write_time", DateTimeUtils.currentTimeMillis()); + c.output(record); + } + })) + + // Step 3. Sink TableRow to BigQuery + .apply("WriteToBigQuery", BigQueryIO.writeTableRows() + .to((SerializableFunction, TableDestination>) input -> { + TableRow row = Objects.requireNonNull(Objects.requireNonNull(input).getValue()); + TableReference tableReference = new TableReference() + .setProjectId("my-project") + .setDatasetId("my-dataset") + .setTableId(row.get("table_id").toString()); + return new TableDestination(tableReference, "Customer names and emails"); + }) + .withoutValidation() + .ignoreUnknownValues() + .withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_NEVER) + .withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_APPEND) + .withMethod(BigQueryIO.Write.Method.STREAMING_INSERTS)); + + pipeline.run(); + } + +}