从Oracle RDBMS迁移至AWS S3（基于Kinesis）：架构与实现问询

Hey Krishnakanth, let’s tackle your Kinesis-based data migration questions with practical, production-grade solutions that fit your multi-downstream consumer needs:

I recommend a layered, modular migration architecture designed to handle both full-load and incremental sync, while supporting multiple downstream consumers:

• Source Data Layer: Your core business database (e.g., MySQL, PostgreSQL, Aurora) where raw data originates.
• Data Capture & Transport Layer:
  • Full-load phase: Use database export tools (AWS DMS full-load mode, mysqldump, pg_dump) to push initial data to Kinesis Streams.
  • Incremental phase: Leverage CDC (Change Data Capture) tools to capture real-time data changes and forward them to dedicated Kinesis Streams per table.
• Stream Processing & Routing Layer:
  • Use Kinesis Data Firehose for batch loading to data warehouses (Redshift) or object storage (S3).
  • Use Kinesis Data Analytics for real-time transformations (e.g., filtering, aggregation) before sending to downstream consumers.
  • Enable Kinesis Enhanced Fan-Out to support up to 50 concurrent consumers per stream without performance hits.
• Target Consumer Layer: Custom Lambda functions, ECS services, or third-party tools that subscribe to streams based on their data needs.

How to Automate Stream Creation for Existing Tables #

Use infrastructure-as-code (IaC) + database metadata scanning to streamline this:

1. Write a script (Python/Shell) that queries your database’s metadata (e.g., INFORMATION_SCHEMA.TABLES for SQL databases) to fetch all tables needing migration.
2. Use AWS CDK (preferred for code-driven infrastructure) to iterate over the table list and create a dedicated Kinesis Stream per table. Example code snippet:

from aws_cdk import aws_kinesis as kinesis
import boto3

def get_table_list_from_db():
    # Custom function to fetch table names from your database
    conn = boto3.client('rds-data') # Adjust for your DB type
    response = conn.execute_statement(
        database='your-db-name',
        resourceArn='your-rds-arn',
        secretArn='your-db-secret-arn',
        sql="SELECT table_name FROM INFORMATION_SCHEMA.TABLES WHERE table_schema = 'public'"
    )
    return [row[0] for row in response['records']]

table_list = get_table_list_from_db()
for table_name in table_list:
    kinesis.Stream(self, f"{table_name}-stream",
                   stream_name=f"{table_name}-data-stream",
                   shard_count=1, # Start small, scale later with CloudWatch metrics
                   retention_period=cdk.Duration.hours(24))

3. Run this IaC deployment before initiating the full-load phase to ensure all streams exist.

Is This Approach Reasonable? #

Absolutely—here’s why:

• Isolation: Dedicated streams prevent cross-table data interference and reduce hot-shard risks for high-volume tables.
• Scalability: Each stream can be independently scaled (adjust shard counts) based on the table’s data throughput.
• Maintainability: Clear table-to-stream mapping simplifies troubleshooting and access control for downstream consumers.

Automating Stream Creation for New Tables #

Set up an event-driven workflow to handle new tables automatically:

1. Capture DDL Events: For AWS-managed databases (RDS/Aurora), use EventBridge to capture CreateTable events. For self-managed databases, create a database trigger that sends a notification (via SNS or direct Lambda call) when a new table is created.
2. Lambda-Based Stream Creation: Write a Lambda function that receives the new table name and creates the corresponding Kinesis Stream via the AWS SDK:

import boto3

kinesis_client = boto3.client('kinesis')

def lambda_handler(event, context):
    table_name = event['detail']['tableName'] # Adjust based on event format
    stream_name = f"{table_name}-data-stream"
    
    # Check if stream already exists to avoid duplicates
    try:
        kinesis_client.describe_stream(StreamName=stream_name)
    except kinesis_client.exceptions.ResourceNotFoundException:
        kinesis_client.create_stream(
            StreamName=stream_name,
            ShardCount=1
        )
    return {"status": "success", "stream_name": stream_name}

3. (Optional) Add AWS Config rules to monitor for unpaired tables/streams and alert on gaps.

Choose a CDC approach based on your database type:

For AWS Managed Databases (RDS/Aurora) #

Use AWS DMS for a fully managed CDC solution:

1. Configure a DMS task with full-load mode first, then enable CDC mode once full load completes.
2. Map each source table to its dedicated Kinesis Stream in the DMS target endpoint configuration.
3. DMS automatically captures INSERT/UPDATE/DELETE events, formats them as JSON (with operation type, old/new values), and pushes them to the corresponding stream.

For Self-Managed Databases #

Use Debezium + Kafka Connect + Kinesis Sink Connector:

1. Deploy Debezium to capture database change logs (e.g., MySQL binlog, PostgreSQL WAL) and send events to a Kafka cluster.
2. Use the Kafka Connect Kinesis Sink Connector to forward these events to the matching Kinesis Streams.

• Alternatively, build a custom CDC script: Monitor database change logs directly, parse events, and use the AWS SDK to push them to Kinesis. Example snippet:

import boto3
import json
from pymysqlreplication import BinLogStreamReader

kinesis_client = boto3.client('kinesis')

def send_to_kinesis(table_name, event_data):
    stream_name = f"{table_name}-data-stream"
    kinesis_client.put_record(
        StreamName=stream_name,
        Data=json.dumps(event_data),
        PartitionKey=event_data['primary_key'] # Use primary key to ensure ordered processing
    )

# Monitor MySQL binlog for changes
stream = BinLogStreamReader(
    connection_settings={"host": "your-db-host", "port": 3306, "user": "cdc-user", "passwd": "your-pass"},
    server_id=100,
    only_events=[WriteRowsEvent, UpdateRowsEvent, DeleteRowsEvent]
)

for binlog_event in stream:
    for row in binlog_event.rows:
        event_payload = {
            "operation": binlog_event.event_type,
            "table": binlog_event.table,
            "data": row,
            "timestamp": binlog_event.timestamp
        }
        send_to_kinesis(binlog_event.table, event_payload)

Multi-Consumer Architecture Tips #

• Enhanced Fan-Out: Enable this feature to let each consumer read streams independently without throttling other consumers.
• Consumer Groups: Use the Kinesis Consumer Library (KCL) for custom consumers to manage shard assignment and avoid duplicate processing across consumer instances.
• Stream Aggregation: For downstream consumers needing cross-table data, use Kinesis Data Analytics to aggregate multiple streams into a single stream for simplified consumption.

内容的提问来源于stack exchange，提问作者Krishnakanth Jc