DynamoDB: Decode CDC event records

Author: amotl

lorrystream/dynamodb_cloud/README.md

@@ -79,6 +79,14 @@ aws dynamodb list-tables
 aws dynamodb describe-table --table-name table-testdrive | grep TableStatus
 ```
 
+### CrateDB Table
+The destination table name in CrateDB is currently hard-coded. Please use
+this command to create the `transactions` table, where the CDC record
+processor will re-materialize CDC events into.
+```shell
+crash -c "CREATE TABLE transactions (data OBJECT(DYNAMIC));"
+```
+
 ### Kinesis Stream
 Capture DynamoDB table operations and relay them to a Kinesis stream.
 ```shell

lorrystream/dynamodb_cloud/backlog.md

@@ -0,0 +1,9 @@
+# DynamoDB CDC processing backlog
+
+## Iteration +1
+- Improve type mapping.
+- Use SQLAlchemy for generating and submitting SQL statement.
+- Improve efficiency by using bulk operations when applicable. 
+
+CREATE TABLE transactions (data OBJECT(DYNAMIC));
+CREATE TABLE transactions (id INT) WITH (column_policy = 'dynamic');

lorrystream/dynamodb_cloud/decoder.py

@@ -0,0 +1,106 @@
+# ruff: noqa: S608
+import json
+import logging
+import typing as t
+from collections import OrderedDict
+
+from lorrystream.util.data import asbool
+
+logger = logging.getLogger(__name__)
+
+
+class OpsLogDecoder:
+    """
+    Utilities for decoding DynamoDB CDC operations events.
+    """
+
+    @classmethod
+    def decode_opslog_item(cls, record: t.Dict[str, t.Any]):
+        """
+        DROP TABLE transactions;
+        CREATE TABLE transactions (id INT) WITH (column_policy = 'dynamic');
+        CREATE TABLE transactions (data OBJECT(DYNAMIC));
+
+        -- https://www.singlestore.com/blog/cdc-data-from-dynamodb-to-singlestore-using-dynamodb-streams/
+        """
+        event_source = record.get("eventSource")
+        event_name = record.get("eventName")
+        if event_source != "aws:dynamodb":
+            raise ValueError(f"Unknown eventSource: {event_source}")
+
+        if event_name == "INSERT":
+            json_str = json.dumps(cls.materialize_new_image(record["dynamodb"]["NewImage"]))
+            sql = f"INSERT INTO transactions (data) VALUES ('{json_str}');".strip()
+
+        elif event_name == "MODIFY":
+            key1 = record["dynamodb"]["Keys"]["device"]["S"]
+            key2 = record["dynamodb"]["Keys"]["timestamp"]["S"]
+            json_str = json.dumps(cls.materialize_new_image(record["dynamodb"]["NewImage"]))
+            sql = f"""
+                UPDATE transactions
+                SET data = '{json_str}'
+                WHERE data['device'] = '{key1}' AND data['timestamp'] = '{key2}';""".strip()
+
+        elif event_name == "REMOVE":
+            key1 = record["dynamodb"]["Keys"]["device"]["S"]
+            key2 = record["dynamodb"]["Keys"]["timestamp"]["S"]
+            sql = f"""
+                DELETE FROM transactions
+                WHERE data['device'] = '{key1}' AND data['timestamp'] = '{key2}';""".strip()
+
+        else:
+            raise ValueError(f"Unknown CDC event name: {event_name}")
+
+        return sql
+
+    @classmethod
+    def materialize_new_image(cls, item: t.Dict[str, t.Dict[str, str]]) -> t.Dict[str, str]:
+        """
+        {
+            "humidity": {"N": "84.84"},
+            "temperature": {"N": "42.42"},
+            "device": {"S": "qux"},
+            "timestamp": {"S": "2024-07-12T01:17:42"},
+        }
+
+        A complete list of DynamoDB data type descriptors:
+
+        S – String
+        N – Number
+        B – Binary
+        BOOL – Boolean
+        NULL – Null
+        M – Map
+        L – List
+        SS – String Set
+        NS – Number Set
+        BS – Binary Set
+
+        """
+        out = OrderedDict()
+        for key, value_composite in item.items():
+            type_: str = list(value_composite.keys())[0]
+            value: t.Any = list(value_composite.values())[0]
+            if type_ == "S":
+                # TODO: Add heuristics for detecting types of timestamps or others?
+                pass
+            elif type_ == "N":
+                value = float(value)
+            elif type_ == "B":
+                raise NotImplementedError(f"Type not implemented yet: {type_}")
+            elif type_ == "BOOL":
+                value = asbool(value)
+            elif type_ == "NULL":
+                value = None
+            elif type_ == "M":
+                raise NotImplementedError(f"Type not implemented yet: {type_}")
+            elif type_ == "L":
+                raise NotImplementedError(f"Type not implemented yet: {type_}")
+            elif type_ == "SS":
+                raise NotImplementedError(f"Type not implemented yet: {type_}")
+            elif type_ == "NS":
+                raise NotImplementedError(f"Type not implemented yet: {type_}")
+            elif type_ == "BS":
+                raise NotImplementedError(f"Type not implemented yet: {type_}")
+            out[key] = value
+        return out

lorrystream/dynamodb_cloud/dynamodb_cdc_processor.py

@@ -5,13 +5,17 @@
 
 from __future__ import print_function
 
+import json
 import logging
 import logging.handlers as handlers
 import time
 import typing as t
 
 from amazon_kclpy import kcl
 from amazon_kclpy.v3 import processor
+from cratedb_toolkit.util import DatabaseAdapter
+
+from lorrystream.dynamodb_cloud.decoder import OpsLogDecoder
 
 # Logger writes to file because stdout is used by MultiLangDaemon
 logger = logging.getLogger(__name__)
@@ -45,6 +49,7 @@ def __init__(self):
         self._largest_seq: t.Tuple[IntOrNone, IntOrNone] = (None, None)
         self._largest_sub_seq = None
         self._last_checkpoint_time = None
+        self.cratedb = DatabaseAdapter(dburi="crate://")
 
     def initialize(self, initialize_input):
         """
@@ -99,18 +104,21 @@ def checkpoint(self, checkpointer, sequence_number=None, sub_sequence_number=Non
 
     def process_record(self, data, partition_key, sequence_number, sub_sequence_number):
         """
-        Called for each record that is passed to process_records.
+        Convert record, which is a DynamoDB CDC event item, into an SQL statement,
+        and submit to downstream database.
 
         :param str data: The blob of data that was contained in the record.
         :param str partition_key: The key associated with this recod.
         :param int sequence_number: The sequence number associated with this record.
         :param int sub_sequence_number: the sub sequence number associated with this record.
         """
-        ####################################
-        # Insert your processing logic here
-        ####################################
+        cdc_event = json.loads(data)
+        logger.info("CDC event: %s", cdc_event)
+
+        sql = OpsLogDecoder.decode_opslog_item(cdc_event)
+        logger.info("SQL: %s", sql)
 
-        logger.info(data.decode("UTF-8"))
+        self.cratedb.run_sql(sql)
 
     def should_update_sequence(self, sequence_number, sub_sequence_number):
         """

lorrystream/dynamodb_cloud/test_decoder.py

@@ -0,0 +1,94 @@
+from lorrystream.dynamodb_cloud.decoder import OpsLogDecoder
+
+MSG_INSERT = {
+    "awsRegion": "us-east-1",
+    "eventID": "b015b5f0-c095-4b50-8ad0-4279aa3d88c6",
+    "eventName": "INSERT",
+    "userIdentity": None,
+    "recordFormat": "application/json",
+    "tableName": "table-testdrive",
+    "dynamodb": {
+        "ApproximateCreationDateTime": 1720740233012995,
+        "Keys": {"device": {"S": "foo"}, "timestamp": {"S": "2024-07-12T01:17:42"}},
+        "NewImage": {
+            "humidity": {"N": "84.84"},
+            "temperature": {"N": "42.42"},
+            "device": {"S": "foo"},
+            "timestamp": {"S": "2024-07-12T01:17:42"},
+        },
+        "SizeBytes": 99,
+        "ApproximateCreationDateTimePrecision": "MICROSECOND",
+    },
+    "eventSource": "aws:dynamodb",
+}
+MSG_MODIFY = {
+    "awsRegion": "us-east-1",
+    "eventID": "24757579-ebfd-480a-956d-a1287d2ef707",
+    "eventName": "MODIFY",
+    "userIdentity": None,
+    "recordFormat": "application/json",
+    "tableName": "table-testdrive",
+    "dynamodb": {
+        "ApproximateCreationDateTime": 1720742302233719,
+        "Keys": {"device": {"S": "foo"}, "timestamp": {"S": "2024-07-12T01:17:42"}},
+        "NewImage": {
+            "humidity": {"N": "84.84"},
+            "temperature": {"N": "55.66"},
+            "device": {"S": "bar"},
+            "timestamp": {"S": "2024-07-12T01:17:42"},
+        },
+        "OldImage": {
+            "humidity": {"N": "84.84"},
+            "temperature": {"N": "42.42"},
+            "device": {"S": "foo"},
+            "timestamp": {"S": "2024-07-12T01:17:42"},
+        },
+        "SizeBytes": 161,
+        "ApproximateCreationDateTimePrecision": "MICROSECOND",
+    },
+    "eventSource": "aws:dynamodb",
+}
+MSG_REMOVE = {
+    "awsRegion": "us-east-1",
+    "eventID": "ff4e68ab-0820-4a0c-80b2-38753e8e00e5",
+    "eventName": "REMOVE",
+    "userIdentity": None,
+    "recordFormat": "application/json",
+    "tableName": "table-testdrive",
+    "dynamodb": {
+        "ApproximateCreationDateTime": 1720742321848352,
+        "Keys": {"device": {"S": "bar"}, "timestamp": {"S": "2024-07-12T01:17:42"}},
+        "OldImage": {
+            "humidity": {"N": "84.84"},
+            "temperature": {"N": "55.66"},
+            "device": {"S": "bar"},
+            "timestamp": {"S": "2024-07-12T01:17:42"},
+        },
+        "SizeBytes": 99,
+        "ApproximateCreationDateTimePrecision": "MICROSECOND",
+    },
+    "eventSource": "aws:dynamodb",
+}
+
+
+def test_decode_insert():
+    assert (
+        OpsLogDecoder.decode_opslog_item(MSG_INSERT) == "INSERT INTO transactions (data) "
+        'VALUES (\'{"humidity": 84.84, "temperature": 42.42, "device": "foo", "timestamp": "2024-07-12T01:17:42"}\');'
+    )
+
+
+def test_decode_modify():
+    assert (
+        OpsLogDecoder.decode_opslog_item(MSG_MODIFY) == "UPDATE transactions\n                "
+        'SET data = \'{"humidity": 84.84, "temperature": 55.66, '
+        '"device": "bar", "timestamp": "2024-07-12T01:17:42"}\'\n                '
+        "WHERE data['device'] = 'foo' AND data['timestamp'] = '2024-07-12T01:17:42';"
+    )
+
+
+def test_decode_remove():
+    assert (
+        OpsLogDecoder.decode_opslog_item(MSG_REMOVE) == "DELETE FROM transactions\n                "
+        "WHERE data['device'] = 'bar' AND data['timestamp'] = '2024-07-12T01:17:42';"
+    )