From Transactional Databases to Iceberg: Building a Multi-Source CDC Lakehouse on AWS

SET CLUSTER SETTING kv.rangefeed.enabled = true;

CREATE CHANGEFEED FOR TABLE defaultdb.financial_transactions
  INTO 'external://msk_financial'
  WITH format = json, updated, resolved = '30s', envelope = 'enriched', key_in_value;

// insert / update / delete
{ "after": {"transaction_id": "TXN-001", "amount": 150.00}, "key": {"transaction_id": "TXN-001"}, "op": "c" }
{ "after": {"transaction_id": "TXN-001", "amount": 200.00}, "key": {"transaction_id": "TXN-001"}, "op": "u" }
{ "after": null,                                            "key": {"transaction_id": "TXN-001"}, "op": "d" }

resource "aws_kinesis_firehose_delivery_stream" "delivery_stream" {
  destination = "iceberg"

  msk_source_configuration {
    msk_cluster_arn = var.MSK_CLUSTER_ARN
    topic_name      = var.TOPIC_NAME
    authentication_configuration {
      connectivity = "PRIVATE"   # PrivateLink into the VPC
      role_arn     = aws_iam_role.firehose_role.arn
    }
  }

  iceberg_configuration {
    catalog_arn        = "arn:aws:glue:...:catalog"
    buffering_size     = var.BUFFERING_SIZE
    buffering_interval = var.BUFFERING_INTERVAL

    destination_table_configuration {
      database_name = var.GLUE_DATABASE_NAME
      table_name    = var.GLUE_TABLE_NAME
      unique_keys   = ["transaction_id"]   # upsert key
    }
  }
}

operation_mapping = {"u": "update", "d": "delete"}
op_code = record.get("op")
operation = operation_mapping.get(op_code, "insert") if op_code else "insert"

if operation == "delete":
    transaction_data = record.get("key")   # 'after' is null on deletes
else:
    transaction_data = record.get("after")

connector.class                  = io.debezium.connector.oracle.OracleConnector
database.hostname                = oracle-host.example.com
database.dbname                  = ORCLCDB
database.pdb.name                = ORCLPDB1
topic.prefix                     = oracle-cdc
table.include.list               = <tables to include>
log.mining.strategy              = online_catalog
snapshot.mode                    = initial
decimal.handling.mode            = precise
value.converter                  = org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable   = false
transforms                       = unwrap
transforms.unwrap.type           = io.debezium.transforms.ExtractNewRecordState
transforms.unwrap.add.fields     = op,source.ts_ms
transforms.unwrap.delete.tombstone.handling.mode = rewrite

public static void main(String[] args) throws Exception {
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    env.enableCheckpointing(60_000);   // commit cadence, tunable; see the latency section

    CatalogLoader glueLoader = CatalogLoader.custom("glue_catalog", Map.of(
        CatalogProperties.CATALOG_IMPL,       "org.apache.iceberg.aws.glue.GlueCatalog",
        CatalogProperties.WAREHOUSE_LOCATION, "s3://your-warehouse/",
        CatalogProperties.FILE_IO_IMPL,       "org.apache.iceberg.aws.s3.S3FileIO"),
        new Configuration(), "org.apache.iceberg.aws.glue.GlueCatalog");

    // S3 Tables loader is the same builder against the Iceberg REST catalog with SigV4 signing
    buildPipeline(env, glueLoader, s3TablesLoader, "your-kafka-topic", "my_db.my_table", "primary_key");
    env.execute("MyCdcJob");
}

DataStream<RowData> stream = env
    .fromSource(source, WatermarkStrategy.noWatermarks(), "kafka-" + kafkaTopic)
    .map(new MyEnvelopeMapper(schema))
    .name("parse-" + kafkaTopic);

FlinkSink.forRowData(stream)
    .tableLoader(TableLoader.fromCatalog(glueLoader, tableId))
    .upsert(true).equalityFieldColumns(List.of(primaryKey)).append();
// a second FlinkSink writes the same stream to the S3 Tables catalog

public RowData map(String message) throws Exception {
    JsonNode node = JSON.readTree(message);
    String op = node.has("__op") ? node.get("__op").asText() : "c";
    RowKind rowKind = switch (op) {
        case "u" -> RowKind.UPDATE_AFTER;
        case "d" -> RowKind.DELETE;
        default  -> RowKind.INSERT;
    };
    GenericRowData row = new GenericRowData(schema.columns().size());
    for (int i = 0; i < schema.columns().size(); i++) {
        Types.NestedField col = schema.columns().get(i);
        row.setField(i, convertJsonToIcebergType(node.get(col.name()), col.type()));
    }
    row.setRowKind(rowKind);
    return row;
}

MERGE INTO analytics_db.customer_orders t
USING staging_db.cdc_batch s ON t.order_id = s.order_id
WHEN MATCHED AND s.op = 'D' THEN DELETE
WHEN MATCHED THEN UPDATE SET t.order_total = s.order_total
WHEN NOT MATCHED THEN INSERT (order_id, order_total) VALUES (s.order_id, s.order_total);