Upgrade to Pro — share decks privately, control downloads, hide ads and more …

Change Data Capture Pipelines With Debezium and...

Change Data Capture Pipelines With Debezium and Kafka Streams

Streams Change data capture (CDC) via Debezium is liberation for your data: By capturing changes from the log files of the database, it enables a wide range of use cases such as reliable microservices data exchange, the creation of audit logs, invalidating caches and much more.

In this talk we're taking CDC to the next level by exploring the benefits of integrating Debezium with streaming queries via Kafka Streams. Come and join us to learn:

How to run low-latency, time-windowed queries on your operational data
How to enrich audit logs with application-provided metadata
How to materialize aggregate views based on multiple change data streams, ensuring transactional boundaries of the source database

We'll also show how to leverage the Quarkus stack for running your Kafka Streams applications on the JVM, as well as natively via GraalVM, many goodies included, such as its live coding feature for instant feedback during development, health checks, metrics and more.

Gunnar Morling

September 03, 2020
Tweet

More Decks by Gunnar Morling

Other Decks in Programming

Transcript

  1. Debezium What's Change Data Capture? Use Cases Kafka Streams with

    Quarkus Supersonic Subatomic Java The Kafka Streams Extension 1 2 3 Debezium + Kafka Streams = Data Enrichment Auditing Expanding Partial Update Events Aggregate View Materialisation
  2. Gunnar Morling Open source software engineer at Red Hat Debezium

    Quarkus Hibernate Spec Lead for Bean Validation 2.0 Other projects: Deptective, MapStruct Java Champion #Debezium @gunnarmorling
  3. @gunnarmorling Postgres MySQL Kafka Connect Kafka Connect Apache Kafka DBZ

    PG DBZ MySQL Search Index ES Connector JDBC Connector ES Connector ISPN Connector Cache Debezium Enabling Zero-Code Data Streaming Pipelines Data Warehouse #Debezium
  4. @gunnarmorling #Debezium Debezium Connectors MySQL Postgres MongoDB SQL Server Cassandra

    (Incubating) Oracle (Incubating) Db2 (Incubating) Future additions: Vitess, MariaDB
  5. { "before": null, "after": { "id": 1004, "first_name": "Anne", "last_name":

    "Kretchmar", "email": "[email protected]" }, "source": { "name": "dbserver1", "server_id": 0, "ts_sec": 0, "file": "mysql-bin.000003", "pos": 154, "row": 0, "snapshot": true, "db": "inventory", "table": "customers" }, "op": "c", "ts_ms": 1486500577691 } Change Event Structure Key: Primary key of table Value: Describing the change event Old row state New row state Metadata @gunnarmorling #Debezium
  6. Log- vs. Query-Based CDC @gunnarmorling Query-Based Log-Based All data changes

    are captured - + No polling delay or overhead - + Transparent to writing applications and models - + Can capture deletes and old record state - + Installation/Configuration + - #Debezium
  7. Debezium What's Change Data Capture? Use Cases 1 2 3

    Kafka Streams with Quarkus Supersonic Subatomic Java The Kafka Streams Extension Debezium + Kafka Streams = Data Enrichment Auditing Expanding Partial Update Events Aggregate View Materialisation
  8. @gunnarmorling Quarkus Supersonic Subatomic Java “ A Kubernetes Native Java

    stack tailored for OpenJDK HotSpot and GraalVM, crafted from the best of breed Java libraries and standards. #Debezium
  9. Quarkus The Kafka Streams Extension Management of topology Health checks

    Dev Mode Support for native binaries via GraalVM @gunnarmorling #Debezium
  10. Debezium What's Change Data Capture? Use Cases 1 3 2

    Debezium + Kafka Streams = Data Enrichment Auditing Expanding Partial Update Events Aggregate View Materialisation Kafka Streams with Quarkus Supersonic Subatomic Java The Kafka Streams Extension
  11. @gunnarmorling Auditing Source DB Kafka Connect Apache Kafka DBZ Customer

    Events CRM Service Id User Use Case tx-1 Bob Create Customer tx-2 Sarah Delete Customer tx-3 Rebecca Update Customer "Transactions" table #Debezium
  12. @gunnarmorling Auditing Source DB Kafka Connect Apache Kafka DBZ Customer

    Events Transactions CRM Service Id User Use Case tx-1 Bob Create Customer tx-2 Sarah Delete Customer tx-3 Rebecca Update Customer "Transactions" table #Debezium
  13. @gunnarmorling Auditing Source DB Kafka Connect Apache Kafka DBZ Customer

    Events Transactions CRM Service Kafka Streams Id User Use Case tx-1 Bob Create Customer tx-2 Sarah Delete Customer tx-3 Rebecca Update Customer "Transactions" table #Debezium
  14. @gunnarmorling Auditing Source DB Kafka Connect Apache Kafka DBZ Customer

    Events Transactions CRM Service Kafka Streams Id User Use Case tx-1 Bob Create Customer tx-2 Sarah Delete Customer tx-3 Rebecca Update Customer "Transactions" table Enriched Customer Events #Debezium
  15. @gunnarmorling Auditing { "before": { "id": 1004, "last_name": "Kretchmar", "email":

    "[email protected]" }, "after": { "id": 1004, "last_name": "Kretchmar", "email": "[email protected]" }, "source": { "name": "dbserver1", "table": "customers", "txId": "tx-3" }, "op": "u", "ts_ms": 1486500577691 } Customers #Debezium
  16. @gunnarmorling { "before": { "id": 1004, "last_name": "Kretchmar", "email": "[email protected]"

    }, "after": { "id": 1004, "last_name": "Kretchmar", "email": "[email protected]" }, "source": { "name": "dbserver1", "table": "customers", "txId": "tx-3" }, "op": "u", "ts_ms": 1486500577691 } { "before": null, "after": { "id": "tx-3", "user": "Rebecca", "use_case": "Update customer" }, "source": { "name": "dbserver1", "table": "transactions", "txId": "tx-3" }, "op": "c", "ts_ms": 1486500577691 } Transactions Customers { "id": "tx-3" } #Debezium
  17. { "id": "tx-3" } { "before": { "id": 1004, "last_name":

    "Kretchmar", "email": "[email protected]" }, "after": { "id": 1004, "last_name": "Kretchmar", "email": "[email protected]" }, "source": { "name": "dbserver1", "table": "customers", "txId": "tx-3" }, "op": "u", "ts_ms": 1486500577691 } Transactions Customers @gunnarmorling { "before": null, "after": { "id": "tx-3", "user": "Rebecca", "use_case": "Update customer" }, "source": { "name": "dbserver1", "table": "transactions", "txId": "tx-3" }, "op": "c", "ts_ms": 1486500577691 } #Debezium
  18. @gunnarmorling { "before": { "id": 1004, "last_name": "Kretchmar", "email": "[email protected]"

    }, "after": { "id": 1004, "last_name": "Kretchmar", "email": "[email protected]" }, "source": { "name": "dbserver1", "table": "customers", "txId": "tx-3", "user": "Rebecca", "use_case": "Update customer" }, "op": "u", "ts_ms": 1486500577691 } Enriched Customers Auditing #Debezium
  19. @gunnarmorling @Override public KeyValue<JsonObject, JsonObject> transform(JsonObject key, JsonObject value) {

    boolean enrichedAllBufferedEvents = enrichAndEmitBufferedEvents(); if (!enrichedAllBufferedEvents) { bufferChangeEvent(key, value); return null; } KeyValue<JsonObject, JsonObject> enriched = enrichWithTxMetaData(key, value); if (enriched == null) { bufferChangeEvent(key, value); } return enriched; } Auditing Non-trivial join implementation no ordering across topics need to buffer change events until TX data available bit.ly/debezium-auditlogs #Debezium
  20. @gunnarmorling Expanding Partial Update Events Examples MongoDB update events ("patch")

    Postgres Replica identity not FULL TOAST-ed columns Cassandra update events MySQL with row image minimal #Debezium { "before": { ... }, "after": { "id": 1004, "first_name": "Dana", "last_name": "Kretchmar", "email": "[email protected]", "biography": "__debezium_unavailable_value" }, "source": { ... }, "op": "u", "ts_ms": 1570448151611 }
  21. @gunnarmorling Expanding Partial Update Events Examples MongoDB update events ("patch")

    Postgres Replica identity not FULL TOAST-ed columns Cassandra update events MySQL with row image minimal #Debezium { "before": { ... }, "after": { "id": 1004, "first_name": "Dana", "last_name": "Kretchmar", "email": "[email protected]", "biography": "__debezium_unavailable_value" }, "source": { ... }, "op": "u", "ts_ms": 1570448151611 }
  22. Expanding Partial Update Events Obtaining missing values from a state

    store @gunnarmorling #Debezium class ToastColumnValueProvider implements ValueTransformerWithKey<JsonObject, JsonObject, JsonObject> private KeyValueStore<JsonObject, String> biographyStore; @Override public void init(ProcessorContext context) { biographyStore = (KeyValueStore<JsonObject, String>) context.getStateStore( TopologyProducer.BIOGRAPHY_STORE); } @Override public JsonObject transform(JsonObject key, JsonObject value) { // ... } }
  23. Expanding Partial Update Events Obtaining missing values from a state

    store @gunnarmorling #Debezium class ToastColumnValueProvider implements ValueTransformerWithKey<JsonObject, JsonObject, JsonObject> private KeyValueStore<JsonObject, String> biographyStore; @Override public void init(ProcessorContext context) { biographyStore = (KeyValueStore<JsonObject, String>) context.getStateStore( TopologyProducer.BIOGRAPHY_STORE); } @Override public JsonObject transform(JsonObject key, JsonObject value) { // ... } }
  24. Expanding Partial Update Events Obtaining missing values from a state

    store @gunnarmorling #Debezium JsonObject payload = value.getJsonObject("payload"); JsonObject newRowState = payload.getJsonObject("after"); String biography = newRowState.getString("biography"); if (isUnavailableValueMarker(biography)) { String currentValue = biographyStore.get(key); newRowState = Json.createObjectBuilder(newRowState) .add("biography", currentValue) .build(); // ... } else { biographyStore.put(key, biography); } return value;
  25. Expanding Partial Update Events Obtaining missing values from a state

    store @gunnarmorling #Debezium JsonObject payload = value.getJsonObject("payload"); JsonObject newRowState = payload.getJsonObject("after"); String biography = newRowState.getString("biography"); if (isUnavailableValueMarker(biography)) { String currentValue = biographyStore.get(key); newRowState = Json.createObjectBuilder(newRowState) .add("biography", currentValue) .build(); // ... } else { biographyStore.put(key, biography); } return value;
  26. Aggregate View Materialization From Multiple Topics to One View @gunnarmorling

    #Debezium PurchaseOrder OrderLine { "purchaseOrderId" : "order-123", "orderDate" : "2020-08-24", "customerId": "customer-123", "orderLines" : [ { "orderLineId" : "orderLine-456", "productId" : "product-789", "quantity" : 2, "price" : 59.99 }, { "orderLineId" : "orderLine-234", "productId" : "product-567", "quantity" : 1, "price" : 14.99 } 1 n
  27. @gunnarmorling #Debezium Aggregate View Materialization Non-Key Joins (KIP-213) KTable<Long, OrderLine>

    orderLines = ...; KTable<Integer, PurchaseOrder> purchaseOrders = ...; KTable<Integer, PurchaseOrderWithLines> purchaseOrdersWithOrderLines = orderLines .join( purchaseOrders, orderLine -> orderLine.purchaseOrderId, (orderLine, purchaseOrder) -> new OrderLineAndPurchaseOrder(orderLine, purchaseOrder)) .groupBy( (orderLineId, lineAndOrder) -> KeyValue.pair(lineAndOrder.purchaseOrder.id, lineAndOrder)) .aggregate( PurchaseOrderWithLines::new, (Integer key, OrderLineAndPurchaseOrder value, PurchaseOrderWithLines agg) -> agg.addLine(value), (Integer key, OrderLineAndPurchaseOrder value, PurchaseOrderWithLines agg) -> agg.removeLine(value) );
  28. @gunnarmorling #Debezium Aggregate View Materialization Non-Key Joins (KIP-213) KTable<Long, OrderLine>

    orderLines = ...; KTable<Integer, PurchaseOrder> purchaseOrders = ...; KTable<Integer, PurchaseOrderWithLines> purchaseOrdersWithOrderLines = orderLines .join( purchaseOrders, orderLine -> orderLine.purchaseOrderId, (orderLine, purchaseOrder) -> new OrderLineAndPurchaseOrder(orderLine, purchaseOrder)) .groupBy( (orderLineId, lineAndOrder) -> KeyValue.pair(lineAndOrder.purchaseOrder.id, lineAndOrder)) .aggregate( PurchaseOrderWithLines::new, (Integer key, OrderLineAndPurchaseOrder value, PurchaseOrderWithLines agg) -> agg.addLine(value), (Integer key, OrderLineAndPurchaseOrder value, PurchaseOrderWithLines agg) -> agg.removeLine(value) );
  29. @gunnarmorling #Debezium Aggregate View Materialization Non-Key Joins (KIP-213) KTable<Long, OrderLine>

    orderLines = ...; KTable<Integer, PurchaseOrder> purchaseOrders = ...; KTable<Integer, PurchaseOrderWithLines> purchaseOrdersWithOrderLines = orderLines .join( purchaseOrders, orderLine -> orderLine.purchaseOrderId, (orderLine, purchaseOrder) -> new OrderLineAndPurchaseOrder(orderLine, purchaseOrder)) .groupBy( (orderLineId, lineAndOrder) -> KeyValue.pair(lineAndOrder.purchaseOrder.id, lineAndOrder)) .aggregate( PurchaseOrderWithLines::new, (Integer key, OrderLineAndPurchaseOrder value, PurchaseOrderWithLines agg) -> agg.addLine(value), (Integer key, OrderLineAndPurchaseOrder value, PurchaseOrderWithLines agg) -> agg.removeLine(value) );
  30. @gunnarmorling #Debezium Aggregate View Materialization Non-Key Joins (KIP-213) KTable<Long, OrderLine>

    orderLines = ...; KTable<Integer, PurchaseOrder> purchaseOrders = ...; KTable<Integer, PurchaseOrderWithLines> purchaseOrdersWithOrderLines = orderLines .join( purchaseOrders, orderLine -> orderLine.purchaseOrderId, (orderLine, purchaseOrder) -> new OrderLineAndPurchaseOrder(orderLine, purchaseOrder)) .groupBy( (orderLineId, lineAndOrder) -> KeyValue.pair(lineAndOrder.purchaseOrder.id, lineAndOrder)) .aggregate( PurchaseOrderWithLines::new, (Integer key, OrderLineAndPurchaseOrder value, PurchaseOrderWithLines agg) -> agg.addLine(value), (Integer key, OrderLineAndPurchaseOrder value, PurchaseOrderWithLines agg) -> agg.removeLine(value) );
  31. @gunnarmorling #Debezium Aggregate View Materialization Non-Key Joins (KIP-213) KTable<Long, OrderLine>

    orderLines = ...; KTable<Integer, PurchaseOrder> purchaseOrders = ...; KTable<Integer, PurchaseOrderWithLines> purchaseOrdersWithOrderLines = orderLines .join( purchaseOrders, orderLine -> orderLine.purchaseOrderId, (orderLine, purchaseOrder) -> new OrderLineAndPurchaseOrder(orderLine, purchaseOrder)) .groupBy( (orderLineId, lineAndOrder) -> KeyValue.pair(lineAndOrder.purchaseOrder.id, lineAndOrder)) .aggregate( PurchaseOrderWithLines::new, (Integer key, OrderLineAndPurchaseOrder value, PurchaseOrderWithLines agg) -> agg.addLine(value), (Integer key, OrderLineAndPurchaseOrder value, PurchaseOrderWithLines agg) -> agg.removeLine(value) );
  32. @gunnarmorling #Debezium Aggregate View Materialization Awareness of Transaction Boundaries TX

    metadata in change events (e.g. dbserver1.inventory.orderline) { "before": null, "after": { ... }, "source": { ... }, "op": "c", "ts_ms": "1580390884335", "transaction": { "id": "571", "total_order": "1", "data_collection_order": "1" } }
  33. Aggregate View Materialization Awareness of Transaction Boundaries Topic with BEGIN/END

    markers Enable consumers to buffer all events of one transaction @gunnarmorling { "transactionId" : "571", "eventType" : "begin transaction", "ts_ms": 1486500577125 } { "transactionId" : "571", "ts_ms": 1486500577691, "eventType" : "end transaction", "eventCount" : [ { "name" : "dbserver1.inventory.order", "count" : 1 }, { "name" : "dbserver1.inventory.orderLine", "count" : 5 } ] } BEGIN END #Debezium
  34. Takeaways Many Use Cases for Debezium and Kafka Streams Data

    enrichment Creating aggregated events Stream queries Interactive query services for legacy databases @gunnarmorling #Debezium
  35. Takeaways Many Use Cases for Debezium and Kafka Streams Data

    enrichment Creating aggregated events Stream queries Interactive query services for legacy databases @gunnarmorling #Debezium Debezium + Kafka Streams =