Skip to content

Latest commit

 

History

History
81 lines (62 loc) · 1.61 KB

README.md

File metadata and controls

81 lines (62 loc) · 1.61 KB
Raw

what?

A rust-based mysql replication client that emits a pubsub message for each changed mysql row.
Replication position state is stored client-side using a disk-based sqlite db for now.

why?

Because debezium is bad to tweak.
Instead of using debezium SingleMessageTransformers, this instead uses an embedded scripting language (rhai) to transfom messages and destination topics.

This enables easier CDC use-cases, for example: mysql -> cdc-rs -> pubsub topic -> bigquery subscription -> CDC-enabled bigquery table.

how?

nix flake

nix run github:docteurklein/cdc-rs -- --help

OCI image

nix run .#oci-image.copyTo -- docker://docker.io/docteurklein/cdc-rs:latest
docker run --rm -it -v $PWD:$PWD -w $PWD docteurklein/cdc-rs:latest --help

kubernetes

kubectl apply -f $(nix build --no-link --print-out-paths .#kube)

example usage

export SOURCE="mysql://replica:$PASSWORD@127.0.0.1:3306"
cdc-rs \
  --server-id 1 \
  --regex '^pim.*\.pim_catalog_product' \
  --state cdc-rs.sqlite \
  --script script.rhai

Where script.rhai contains:

fn debezium(db, table, changeType, before, after) {
	let ts = timestamp();
	#{
		payload: #{
			before: before,
			after: after,
			op: switch changeType {
				"INSERT" => "c",
				"UPDATE" => "u",
				"DELETE" => "d",
				_ => changeType,
			},
			source: #{
				ts_ms: `${ts}`,
				db: db,
				table: table,
			},
			ts_ms: `${ts}`,
		}
	}
}

fn transform(db, table, changeType, before, after) {
	before.tenant = db;
	after.tenant = db;
	debezium(db, table, changeType, before, after)
}

fn topic(db, table) {
  `prefix.${table}`
}