RDS MariaDb至Kinesis的DMS CDC任务无报错但无数据流入

我尝试通过CDC将RDS MariaDb中的数据同步至Kinesis Stream。DMS复制任务显示执行成功，日志中未发现任何错误，但Kinesis Stream始终未接收到记录。我已遵循AWS官方文档配置。

复制任务配置 #

{
    "Logging": {
        "EnableLogging": true,
        "EnableLogContext": true,
        "LogComponents": [
            {
                "Severity": "LOGGER_SEVERITY_DEFAULT",
                "Id": "TRANSFORMATION"
            },
            {
                "Severity": "LOGGER_SEVERITY_DEBUG",
                "Id": "SOURCE_UNLOAD"
            },
            {
                "Severity": "LOGGER_SEVERITY_DEFAULT",
                "Id": "IO"
            },
            {
                "Severity": "LOGGER_SEVERITY_DEBUG",
                "Id": "TARGET_LOAD"
            },
            {
                "Severity": "LOGGER_SEVERITY_DEFAULT",
                "Id": "PERFORMANCE"
            },
            {
                "Severity": "LOGGER_SEVERITY_DEBUG",
                "Id": "SOURCE_CAPTURE"
            },
            {
                "Severity": "LOGGER_SEVERITY_DEFAULT",
                "Id": "SORTER"
            },
            {
                "Severity": "LOGGER_SEVERITY_DEFAULT",
                "Id": "REST_SERVER"
            },
            {
                "Severity": "LOGGER_SEVERITY_DEFAULT",
                "Id": "VALIDATOR_EXT"
            },
            {
                "Severity": "LOGGER_SEVERITY_DEBUG",
                "Id": "TARGET_APPLY"
            },
            {
                "Severity": "LOGGER_SEVERITY_DEBUG",
                "Id": "TASK_MANAGER"
            },
            {
                "Severity": "LOGGER_SEVERITY_DEFAULT",
                "Id": "TABLES_MANAGER"
            },
            {
                "Severity": "LOGGER_SEVERITY_DEFAULT",
                "Id": "METADATA_MANAGER"
            },
            {
                "Severity": "LOGGER_SEVERITY_DEFAULT",
                "Id": "FILE_FACTORY"
            },
            {
                "Severity": "LOGGER_SEVERITY_DEFAULT",
                "Id": "COMMON"
            },
            {
                "Severity": "LOGGER_SEVERITY_DEFAULT",
                "Id": "ADDONS"
            },
            {
                "Severity": "LOGGER_SEVERITY_DEFAULT",
                "Id": "DATA_STRUCTURE"
            },
            {
                "Severity": "LOGGER_SEVERITY_DEFAULT",
                "Id": "COMMUNICATION"
            },
            {
                "Severity": "LOGGER_SEVERITY_DEFAULT",
                "Id": "FILE_TRANSFER"
            }
        ],
        "CloudWatchLogGroup": "dms-tasks-data-exporter-dev",
        "CloudWatchLogStream": "dms-task-3ODT6664IUAQLWF5J7SV3YZHK7TWAAGBJTVB3SY"
    },
    "StreamBufferSettings": {
        "StreamBufferCount": 3,
        "CtrlStreamBufferSizeInMB": 5,
        "StreamBufferSizeInMB": 8
    },
    "ErrorBehavior": {
        "FailOnNoTablesCaptured": true,
        "ApplyErrorUpdatePolicy": "LOG_ERROR",
        "FailOnTransactionConsistencyBreached": false,
        "RecoverableErrorThrottlingMax": 1800,
        "DataErrorEscalationPolicy": "SUSPEND_TABLE",
        "ApplyErrorEscalationCount": 0,
        "RecoverableErrorStopRetryAfterThrottlingMax": true,
        "RecoverableErrorThrottling": true,
        "ApplyErrorFailOnTruncationDdl": false,
        "DataTruncationErrorPolicy": "LOG_ERROR",
        "ApplyErrorInsertPolicy": "LOG_ERROR",
        "EventErrorPolicy": "IGNORE",
        "ApplyErrorEscalationPolicy": "LOG_ERROR",
        "RecoverableErrorCount": 0,
        "DataErrorEscalationCount": 50,
        "TableErrorEscalationPolicy": "STOP_TASK",
        "RecoverableErrorInterval": 5,
        "ApplyErrorDeletePolicy": "IGNORE_RECORD",
        "TableErrorEscalationCount": 50,
        "FullLoadIgnoreConflicts": true,
        "DataErrorPolicy": "LOG_ERROR",
        "TableErrorPolicy": "SUSPEND_TABLE"
    },
    "TTSettings": {
        "TTS3Settings": null,
        "TTRecordSettings": null,
        "EnableTT": false
    },
    "FullLoadSettings": {
        "CommitRate": 10000,
        "StopTaskCachedChangesApplied": false,
        "StopTaskCachedChangesNotApplied": false,
        "MaxFullLoadSubTasks": 8,
        "TransactionConsistencyTimeout": 600,
        "CreatePkAfterFullLoad": false,
        "TargetTablePrepMode": "DO_NOTHING"
    },
    "TargetMetadata": {
        "ParallelApplyBufferSize": 100,
        "ParallelApplyQueuesPerThread": 1,
        "ParallelApplyThreads": 0,
        "TargetSchema": "",
        "InlineLobMaxSize": 0,
        "ParallelLoadQueuesPerThread": 1,
        "SupportLobs": true,
        "LobChunkSize": 64,
        "TaskRecoveryTableEnabled": false,
        "ParallelLoadThreads": 0,
        "LobMaxSize": 32,
        "BatchApplyEnabled": false,
        "FullLobMode": false,
        "LimitedSizeLobMode": true,
        "LoadMaxFileSize": 0,
        "ParallelLoadBufferSize": 0
    },
    "BeforeImageSettings": {
        "EnableBeforeImage": false,
        "ColumnFilter": "pk-only",
        "FieldName": ""
    },
    "ControlTablesSettings": {
        "historyTimeslotInMinutes": 5,
        "HistoryTimeslotInMinutes": 5,
        "StatusTableEnabled": false,
        "SuspendedTablesTableEnabled": false,
        "HistoryTableEnabled": false,
        "ControlSchema": "",
        "FullLoadExceptionTableEnabled": false
    },
    "LoopbackPreventionSettings": null,
    "CharacterSetSettings": null,
    "FailTaskWhenCleanTaskResourceFailed": false,
    "ChangeProcessingTuning": {
        "StatementCacheSize": 50,
        "CommitTimeout": 1,
        "BatchApplyPreserveTransaction": true,
        "BatchApplyTimeoutMin": 1,
        "BatchSplitSize": 0,
        "BatchApplyTimeoutMax": 30,
        "MinTransactionSize": 1000,
        "MemoryKeepTime": 60,
        "BatchApplyMemoryLimit": 500,
        "MemoryLimitTotal": 1024
    },
    "ChangeProcessingDdlHandlingPolicy": {
        "HandleSourceTableDropped": false,
        "HandleSourceTableTruncated": false,
        "HandleSourceTableAltered": false
    },
    "PostProcessingRules": null
}

表映射 #

{
    "rules": [
        {
            "rule-type": "selection",
            "rule-id": "1",
            "rule-name": "1",
            "rule-action": "include",
            "object-locator": {
                "schema-name": "fosanis_cdc_test",
                "table-name": "%"
            }
        },
        {
            "rule-type": "object-mapping",
            "rule-id": "2",
            "rule-name": "DefaultMapToKinesis",
            "rule-action": "map-record-to-record",
            "object-locator": {
                "schema-name": "fosanis_cdc_test",
                "table-name": "customers"
            }
        },
        {
            "rule-type": "object-mapping",
            "rule-id": "3",
            "rule-name": "DefaultMapToKinesis",
            "rule-action": "map-record-to-record",
            "object-locator": {
                "schema-name": "fosanis_cdc_test",
                "table-name": "contacts"
            }
        }
    ]
}

Kinesis端点配置 #

{
    "StreamArn": "arn:aws:kinesis:eu-central-1:614797193252:stream/data_exporter_dev",
    "MessageFormat": "json-unformatted",
    "ServiceAccessRoleArn": "arn:aws:iam::614797193252:role/data_exporter_kinesis_dev",
    "IncludeTransactionDetails": true,
    "IncludePartitionValue": true,
    "PartitionIncludeSchemaTable": true,
    "IncludeTableAlterOperations": true,
    "IncludeControlDetails": true,
    "IncludeNullAndEmpty": false
}

补充说明 #

该复制任务同步数据至S3存储桶时可正常工作，对应的S3配置如下：

{
    "ServiceAccessRoleArn": "arn:aws:iam::614797193252:role/S3AdminForDms",
    "CsvRowDelimiter": "\\n",
    "CsvDelimiter": ",",
    "BucketName": "mika-dms-test",
    "CompressionType": "NONE",
    "EnableStatistics": true,
    "DatePartitionEnabled": false
}

额外连接属性为：bucketName=mika-dms-test;compressionType=NONE;csvDelimiter=,;csvRowDelimiter=\n;datePartitionEnabled=false;

排查建议 #

• 验证Kinesis IAM角色权限：确认data_exporter_kinesis_dev角色拥有kinesis:PutRecord和kinesis:PutRecords权限，且信任策略允许dms.amazonaws.com作为可信实体。
• 检查DMS任务数据捕获情况：查看任务的「表统计信息」，确认是否有捕获到Insert/Update/Delete记录。如果是全量加载模式，确认任务已完成全量阶段且有数据输出；如果仅CDC模式，需确保源表有新的变更操作。
• 检查表映射规则有效性：尝试简化表映射规则，仅保留初始的选择规则（去掉后续的object-mapping规则），测试是否能正常同步数据，排除规则冲突的可能。
• 提升DMS日志详细度：将TARGET_APPLY和SOURCE_CAPTURE的日志级别调整为TRACE，查看更细粒度的写入Kinesis过程日志，排查是否存在静默失败。
• 确认Kinesis Stream状态：检查Stream是否处于ACTIVE状态，分片数量是否满足写入需求，是否存在限流情况（可查看CloudWatch指标PutRecords.Success）。

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