기계 번역으로 제공되는 번역입니다. 제공된 번역과 원본 영어의 내용이 상충하는 경우에는 영어 버전이 우선합니다.

아파치 카프카를 타겟으로 사용 AWS Database Migration Service

를 사용하여 데이터를 Apache AWS DMS Kafka 클러스터로 마이그레이션할 수 있습니다. Apache Kafka는 분산 스트리밍 플랫폼입니다. Apache Kafka를 사용하여 스트리밍 데이터의 실시간 수집 및 처리를 수행할 수 있습니다.

AWS 또한 타겟으로 사용할 수 있는 Apache Kafka용 아마존 매니지드 스트리밍 (Amazon MSK) 도 제공합니다. AWS DMS Amazon MSK는 Apache Kafka 인스턴스의 구현 및 관리를 단순화하는 완전 관리형 Apache Kafka 스트리밍 서비스입니다. 오픈 소스 Apache Kafka 버전에서 작동하며, 다른 Apache Kafka 인스턴스와 마찬가지로 Amazon MSK 인스턴스에 AWS DMS 대상으로 액세스할 수 있습니다. 자세한 내용은 Amazon Managed Streaming for Apache Kafka 개발자 안내서의 Amazon MSK란 무엇인가요?를 참조하세요.

Kafka 클러스터는 여러 파티션으로 구분된 주제라는 범주에 레코드 스트림을 저장합니다. 파티션은 주제에서 고유하게 식별되는 데이터 레코드(메시지) 시퀀스입니다. 파티션은 주제 레코드의 병렬 처리가 가능하도록 한 클러스터의 여러 브로커로 분산될 수 있습니다. Apache Kafka의 주제 및 파티션과 분산에 관한 자세한 내용은 주제 및 로그와 분산을 참조하십시오.

Kafka 클러스터는 Amazon MSK 인스턴스, Amazon EC2 인스턴스에서 실행되는 클러스터 또는 온프레미스 클러스터일 수 있습니다. Amazon MSK 인스턴스 또는 Amazon EC2 인스턴스의 클러스터는 동일한 VPC에 있을 수도 있고 다른 VPC에 있을 수도 있습니다. 클러스터가 온프레미스인 경우 복제 인스턴스용 자체 온프레미스 이름 서버를 사용하여 클러스터의 호스트 이름을 확인할 수 있습니다. 복제 인스턴스에서 이름 서버를 설정하는 방법에 관한 자세한 내용은 자체 온프레미스 이름 서버 사용 단원을 참조하십시오. 네트워크 설정에 관한 자세한 내용은 복제 인스턴스용으로 네트워크 설정 단원을 참조하십시오.

Amazon MSK 클러스터를 사용할 때는 해당 보안 그룹이 복제 인스턴스에서의 액세스를 허용하는지 확인하십시오. Amazon MSK 클러스터의 보안 그룹 변경에 관한 자세한 내용은 Amazon MSK 클러스터의 보안 그룹 변경을 참조하십시오.

AWS Database Migration Service JSON을 사용하여 Kafka 주제에 레코드를 게시합니다. 변환 중 AWS DMS 는 각 레코드를 소스 데이터베이스로부터 JSON 형식의 속성-값 페어로 직렬화합니다.

지원되는 데이터 소스에서 대상 Kafka 클러스터로 데이터를 마이그레이션하려면 객체 매핑을 사용합니다. 객체 매핑을 통해 데이터 레코드를 대상 주제에 구조화하는 방법을 결정합니다. 또한 각 테이블에 대한 파티션 키를 정의합니다. Apache Kafka가 이러한 테이블을 사용해 데이터를 파티션으로 그룹화합니다.

현재는 작업당 단일 주제를 AWS DMS 지원합니다. 테이블이 여러 개 있는 단일 작업의 경우 모든 메시지가 단일 주제로 이동합니다. 각 메시지에는 대상 스키마와 테이블을 식별하는 메타데이터 섹션이 포함되어 있습니다. AWS DMS 버전 3.4.6 이상에서는 객체 매핑을 사용한 멀티토픽 복제를 지원합니다. 자세한 설명은 객체 매핑을 사용한 멀티주제 복제 섹션을 참조하세요.

ParallelApply* 작업 설정을 사용할 때 partition-key-type 기본값은 테이블의 schema-name.table-name가 아니라 primary-key입니다.

전송 계층 보안(TLS)을 사용하여 Kafka에 연결

Kafka 클러스터는 전송 계층 보안(TLS)을 사용한 보안 연결만 허용합니다. DMS를 사용하면 다음 세 가지 보안 프로토콜 옵션 중 하나를 사용하여 Kafka 엔드포인트 연결을 보호할 수 있습니다.

Amazon MSK 또는 자체 관리형 Kafka 클러스터에서 SSL 암호화 사용

Amazon MSK 또는 자체 관리형 Kafka 클러스터에 대한 엔드포인트 연결을 보호하는 데 SSL 암호화를 사용할 수 있습니다. SSL 암호화 인증 방법을 사용하면 클라이언트가 서버의 인증서를 통해 서버의 ID를 검증합니다. 그러면 서버와 클라이언트 간에 암호화된 연결이 이루어집니다.

"KafkaSettings": {
    "SecurityProtocol": "ssl-encryption", 
}
                

SSL 인증 사용

Amazon MSK 또는 자체 관리형 Kafka 클러스터에 대한 엔드포인트 연결을 보호하는 데 SSL 인증을 사용할 수 있습니다.

Amazon MSK에 연결할 때 SSL 인증을 사용한 클라이언트 인증 및 암호화를 활성화하려면 다음을 수행하십시오.

SASL-SSL 인증을 사용하여 Amazon MSK에 연결

SASL(Simple Authentication and Security Layer) 메서드는 사용자 이름과 암호를 사용하여 클라이언트 ID를 검증하고 서버와 클라이언트 간에 암호화된 연결을 만듭니다.

SASL을 사용하려면 Amazon MSK 클러스터를 설정할 때 먼저 안전한 사용자 이름과 암호를 생성해야 합니다. Amazon MSK 클러스터의 보안 사용자 이름과 암호를 설정하는 방법에 관한 설명은 Amazon Managed Streaming for Apache Kafka 개발자 안내서의 Amazon MSK 클러스터용 SASL/SCRAM 인증 설정을 참조하세요.

그런 다음, Kafka 대상 엔드포인트를 생성할 때 sasl-ssl 옵션을 사용하여 보안 프로토콜 엔드포인트 설정(SecurityProtocol)을 설정합니다. SaslUsername 및 SaslPassword 옵션도 설정합니다. Amazon MSK 클러스터를 처음 설정할 때 다음 JSON 예제에서 보는 것처럼 이 옵션 설정이 보안 사용자 이름 및 암호와 일치하는지 확인하십시오.

                   
"KafkaSettings": {
    "SecurityProtocol": "sasl-ssl",
    "SaslUsername":"Amazon MSK cluster secure user name",
    "SaslPassword":"Amazon MSK cluster secure password"                    
}
                    

Apache Kafka 대상의 경우 이전 이미지를 사용하여 CDC 행의 원래 값 보기

Kafka 같은 데이터 스트리밍 대상에 CDC 업데이트를 작성하는 경우 업데이트를 통한 변경 전에 소스 데이터베이스 행의 원래 값을 볼 수 있습니다. 이를 가능하게 하기 위해 소스 데이터베이스 엔진에서 제공한 데이터를 기반으로 업데이트 이벤트의 이전 이미지를 AWS DMS 채웁니다.

소스 데이터베이스 엔진에 따라 이전 이미지에 대해 서로 다른 양의 정보를 제공합니다.

이전 이미징을 활성화하여 소스 데이터베이스의 원래 값을 AWS DMS 출력에 추가하려면 BeforeImageSettings 태스크 설정 또는 add-before-image-columns 파라미터를 사용합니다. 이 파라미터는 열 변환 규칙을 적용합니다.

BeforeImageSettings는 다음과 같이 소스 데이터베이스 시스템에서 수집된 값을 사용하여 모든 업데이트 작업에 새 JSON 속성을 추가합니다.

"BeforeImageSettings": {
    "EnableBeforeImage": boolean,
    "FieldName": string,  
    "ColumnFilter": pk-only (default) / non-lob / all (but only one)
}
        

BeforeImageSettings 옵션의 경우, 다음이 적용됩니다.

이전 이미지 변환 규칙 사용

작업 설정 대신 열 변환 규칙을 적용하는 add-before-image-columns 파라미터를 사용할 수 있습니다. 이 파라미터를 사용하면 Kafka 같은 데이터 스트리밍 대상에서 CDC 중에 이전 이미징을 활성화할 수 있습니다.

변환 규칙에 add-before-image-columns를 사용하면 이전 이미지 결과를 보다 세밀하게 제어할 수 있습니다. 변환 규칙을 통해 객체 로케이터를 사용하여 규칙에 대해 선택한 테이블을 제어할 수 있습니다. 또한 변환 규칙을 함께 연결하여 테이블마다 서로 다른 규칙을 적용할 수 있습니다. 그런 다음, 다른 규칙을 사용하여 생성된 열을 조작할 수 있습니다.

해당 열에 대해 add-before-image-columns 파라미터가 있는 transformation 규칙 유형이 before-image-def 단원을 제공해야 합니다. 다음은 그 한 예입니다.

    {
      "rule-type": "transformation",
      …
      "rule-target": "column",
      "rule-action": "add-before-image-columns",
      "before-image-def":{
        "column-filter": one-of  (pk-only / non-lob / all),
        "column-prefix": string,
        "column-suffix": string,
      }
    }

column-prefix의 값은 열 이름 앞에 추가되며, column-prefix의 기본값은 BI_입니다. column-suffix의 값은 열 이름 뒤에 추가되며, 기본값은 비어 있습니다. column-prefix와 column-suffix를 모두 빈 문자열로 설정하지 마십시오.

column-filter에 대해 하나의 값을 선택합니다. 테이블 기본 키의 일부인 열만 추가하려면 pk-only를 선택하고, LOB 유형이 아닌 열만 추가하려면 non-lob를 선택하고, 이전 이미지 값이 있는 모든 열을 추가하려면 all을 선택합니다.

이전 이미지 변환 규칙의 예

다음 예의 변환 규칙은 대상에서 BI_emp_no라는 새 열을 추가합니다. UPDATE employees SET emp_no = 3 WHERE emp_no = 1; 같은 문은 BI_emp_no 필드를 1로 채웁니다. Amazon S3 대상에 CDC 업데이트를 작성할 때 BI_emp_no 열을 통해 업데이트된 원래 행을 알 수 있습니다.

{
  "rules": [
    {
      "rule-type": "selection",
      "rule-id": "1",
      "rule-name": "1",
      "object-locator": {
        "schema-name": "%",
        "table-name": "%"
      },
      "rule-action": "include"
    },
    {
      "rule-type": "transformation",
      "rule-id": "2",
      "rule-name": "2",
      "rule-target": "column",
      "object-locator": {
        "schema-name": "%",
        "table-name": "employees"
      },
      "rule-action": "add-before-image-columns",
      "before-image-def": {
        "column-prefix": "BI_",
        "column-suffix": "",
        "column-filter": "pk-only"
      }
    }
  ]
}

add-before-image-columns 규칙 작업 사용에 관한 자세한 내용은 변환 규칙 및 작업 단원을 참조하십시오.

Apache Kafka를 대상으로 사용할 때의 제한 사항 AWS Database Migration Service

Apache Kafka를 대상으로 사용할 때 다음 제한 사항이 적용됩니다.

객체 매핑을 사용하여 데이터를 Kafka 주제로 마이그레이션

AWS DMS 테이블 매핑 규칙을 사용하여 원본의 데이터를 대상 Kafka 주제에 매핑합니다. 데이터를 대상 주제에 매핑하려면 객체 매핑이라는 테이블 매핑 규칙 유형을 사용합니다. 객체 매핑을 사용하여 원본의 데이터 레코드를 Kafka 주제에 게시된 데이터 레코드로 매핑하는 방법을 지정합니다.

Kafka 주제에는 파티션 키 외에 별다른 사전 설정 구조가 없습니다.

object-mapping 규칙을 만들려면 rule-type을 object-mapping으로 지정합니다. 이 규칙은 사용할 객체 매핑의 유형을 지정합니다.

이 규칙의 구조는 다음과 같습니다.

{
    "rules": [
        {
            "rule-type": "object-mapping",
            "rule-id": "id",
            "rule-name": "name",
            "rule-action": "valid object-mapping rule action",
            "object-locator": {
                "schema-name": "case-sensitive schema name",
                "table-name": ""
            }
        }
    ]
}

AWS DMS 현재 매개 변수에 대해 map-record-to-record 및 map-record-to-document 만 유효한 값으로 지원합니다. rule-action 이러한 설정은 exclude-columns 속성 목록의 일부로 제외되지 않은 값에 영향을 줍니다. map-record-to-record및 map-record-to-document 값은 이러한 레코드를 기본적으로 AWS DMS 처리하는 방법을 지정합니다. 이러한 값은 어떤 식으로든 속성 매핑에 영향을 미치지 않습니다.

관계형 데이터베이스에서 Kafka 주제로 마이그레이션할 때 map-record-to-record를 사용합니다. 이러한 규칙 유형은 Kafka 주제의 파티션 키로 관계형 데이터베이스의 taskResourceId.schemaName.tableName 값을 사용하며 소스 데이터베이스의 각 열마다 속성을 하나씩 생성합니다.

map-record-to-record을 사용하는 경우 다음 사항에 유의하세요.

map-record-to-record를 이해하는 한 가지 방법은 작업 중일 때 관찰하는 것입니다. 이 예에서는 다음 구조와 데이터를 사용하여 관계형 데이터베이스 테이블 행에서 시작한다고 가정합니다.

이 정보를 Test라는 스키마에서 Kafka 주제로 마이그레이션하려면 데이터를 대상 주제로 매핑하는 규칙을 생성합니다. 다음 규칙은 그 매핑 과정을 보여 줍니다.

{
    "rules": [
        {
            "rule-type": "selection",
            "rule-id": "1",
            "rule-name": "1",
            "rule-action": "include",
            "object-locator": {
                "schema-name": "Test",
                "table-name": "%"
            }
        },
        {
            "rule-type": "object-mapping",
            "rule-id": "2",
            "rule-name": "DefaultMapToKafka",
            "rule-action": "map-record-to-record",
            "object-locator": {
                "schema-name": "Test",
                "table-name": "Customers"
            }
        }
    ]
}

다음은 지정된 Kafka 주제와 파티션 키(이 경우 taskResourceId.schemaName.tableName)로 Kafka 대상 주제의 샘플 데이터를 사용하여 결과 레코드 형식을 보여줍니다.

  {
     "FirstName": "Randy",
     "LastName": "Marsh",
     "StoreId":  "5",
     "HomeAddress": "221B Baker Street",
     "HomePhone": "1234567890",
     "WorkAddress": "31 Spooner Street, Quahog",
     "WorkPhone": "9876543210",
     "DateOfBirth": "02/29/1988"
  }              

속성 매핑으로 날짜 재구성

속성 맵을 사용하여 데이터를 Kafka 주제로 마이그레이션하는 동안 데이터를 재구성할 수 있습니다. 예를 들어 소스의 필드 몇 개를 대상의 단일 필드로 묶어야 하는 경우도 있을 것입니다. 다음의 속성 맵은 날짜를 재구성하는 방법을 보여줍니다.

{
    "rules": [
        {
            "rule-type": "selection",
            "rule-id": "1",
            "rule-name": "1",
            "rule-action": "include",
            "object-locator": {
                "schema-name": "Test",
                "table-name": "%"
            }
        },
        {
            "rule-type": "object-mapping",
            "rule-id": "2",
            "rule-name": "TransformToKafka",
            "rule-action": "map-record-to-record",
            "target-table-name": "CustomerData",
            "object-locator": {
                "schema-name": "Test",
                "table-name": "Customers"
            },
            "mapping-parameters": {
                "partition-key-type": "attribute-name",
                "partition-key-name": "CustomerName",
                "exclude-columns": [
                    "firstname",
                    "lastname",
                    "homeaddress",
                    "homephone",
                    "workaddress",
                    "workphone"
                ],
                "attribute-mappings": [
                    {
                        "target-attribute-name": "CustomerName",
                        "attribute-type": "scalar",
                        "attribute-sub-type": "string",
                        "value": "${lastname}, ${firstname}"
                    },
                    {
                        "target-attribute-name": "ContactDetails",
                        "attribute-type": "document",
                        "attribute-sub-type": "json",
                        "value": {
                            "Home": {
                                "Address": "${homeaddress}",
                                "Phone": "${homephone}"
                            },
                            "Work": {
                                "Address": "${workaddress}",
                                "Phone": "${workphone}"
                            }
                        }
                    }
                ]
            }
        }
    ]
}

partition-key에 상수 값을 설정하려면 partition-key 값을 지정합니다. 예를 들어 이 방법을 통해 모든 데이터를 단일 파티션에 저장할 수 있습니다. 다음 매핑은 이러한 접근 방식을 보여줍니다.

{
    "rules": [
        {
            "rule-type": "selection",
            "rule-id": "1",
            "rule-name": "1",
            "object-locator": {
                "schema-name": "Test",
                "table-name": "%"
            },
            "rule-action": "include"
        },
        {
            "rule-type": "object-mapping",
            "rule-id": "1",
            "rule-name": "TransformToKafka",
            "rule-action": "map-record-to-document",
            "object-locator": {
                "schema-name": "Test",
                "table-name": "Customer"
            },
            "mapping-parameters": {
                "partition-key": {
                    "value": "ConstantPartitionKey"
                },
                "exclude-columns": [
                    "FirstName",
                    "LastName",
                    "HomeAddress",
                    "HomePhone",
                    "WorkAddress",
                    "WorkPhone"
                ],
                "attribute-mappings": [
                    {
                        "attribute-name": "CustomerName",
                        "value": "${FirstName},${LastName}"
                    },
                    {
                        "attribute-name": "ContactDetails",
                        "value": {
                            "Home": {
                                "Address": "${HomeAddress}",
                                "Phone": "${HomePhone}"
                            },
                            "Work": {
                                "Address": "${WorkAddress}",
                                "Phone": "${WorkPhone}"
                            }
                        }
                    },
                    {
                        "attribute-name": "DateOfBirth",
                        "value": "${DateOfBirth}"
                    }
                ]
            }
        }
    ]
}

객체 매핑을 사용한 멀티주제 복제

기본적으로 AWS DMS 작업은 모든 소스 데이터를 다음 Kafka 주제 중 하나로 마이그레이션합니다.

AWS DMS 엔진 버전 3.4.6 이상에서는 kafka-target-topic 속성을 사용하여 마이그레이션된 각 소스 테이블을 별도의 주제에 매핑할 수 있습니다. 예를 들어, 다음 객체 매핑 규칙은 원본 테이블 Customer과 Address를 Kafka 주제 customer_topic 및 address_topic에 각각 마이그레이션합니다. 동시에 Test 스키마의 테이블을 포함하여 다른 모든 원본 테이블을 대상 Bills 엔드포인트에 지정된 주제로 AWS DMS 마이그레이션합니다.

{
    "rules": [
        {
            "rule-type": "selection",
            "rule-id": "1",
            "rule-name": "1",
            "rule-action": "include",
            "object-locator": {
                "schema-name": "Test",
                "table-name": "%"
            }
        },
        {
            "rule-type": "object-mapping",
            "rule-id": "2",
            "rule-name": "MapToKafka1",
            "rule-action": "map-record-to-record",
            "kafka-target-topic": "customer_topic",
            "object-locator": {
                "schema-name": "Test",
                "table-name": "Customer" 
            },
            "partition-key": {"value": "ConstantPartitionKey" }
        },
        {
            "rule-type": "object-mapping",
            "rule-id": "3",
            "rule-name": "MapToKafka2",
            "rule-action": "map-record-to-record",
            "kafka-target-topic": "address_topic",
            "object-locator": {
                "schema-name": "Test",
                "table-name": "Address"
            },
            "partition-key": {"value": "HomeAddress" }
        },
        {
            "rule-type": "object-mapping",
            "rule-id": "4",
            "rule-name": "DefaultMapToKafka",
            "rule-action": "map-record-to-record",
            "object-locator": {
                "schema-name": "Test",
                "table-name": "Bills"
            }
        }
    ]
}

Kafka 멀티주제 복제를 사용하면 단일 복제 작업을 사용하여 원본 테이블을 그룹화하여 별도의 Kafka 주제로 마이그레이션할 수 있습니다.

Apache Kafka 메시지 형식

JSON 출력은 단지 키-값 페어의 목록입니다.

다음 JSON 메시지 예제는 모든 추가 메타데이터가 포함된 데이터 유형 메시지를 보여줍니다.

{ 
   "data":{ 
      "id":100000161,
      "fname":"val61s",
      "lname":"val61s",
      "REGION":"val61s"
   },
   "metadata":{ 
      "timestamp":"2019-10-31T22:53:59.721201Z",
      "record-type":"data",
      "operation":"insert",
      "partition-key-type":"primary-key",
      "partition-key-value":"sbtest.sbtest_x.100000161",
      "schema-name":"sbtest",
      "table-name":"sbtest_x",
      "transaction-id":9324410911751,
      "transaction-record-id":1,
      "prev-transaction-id":9324410910341,
      "prev-transaction-record-id":10,
      "commit-timestamp":"2019-10-31T22:53:55.000000Z",
      "stream-position":"mysql-bin-changelog.002171:36912271:0:36912333:9324410911751:mysql-bin-changelog.002171:36912209"
   }
}
                

다음 JSON 메시지 예제는 컨트롤 유형 메시지를 보여줍니다.

{ 
   "control":{ 
      "table-def":{ 
         "columns":{ 
            "id":{ 
               "type":"WSTRING",
               "length":512,
               "nullable":false
            },
            "fname":{ 
               "type":"WSTRING",
               "length":255,
               "nullable":true
            },
            "lname":{ 
               "type":"WSTRING",
               "length":255,
               "nullable":true
            },
            "REGION":{ 
               "type":"WSTRING",
               "length":1000,
               "nullable":true
            }
         },
         "primary-key":[ 
            "id"
         ],
         "collation-name":"latin1_swedish_ci"
      }
   },
   "metadata":{ 
      "timestamp":"2019-11-21T19:14:22.223792Z",
      "record-type":"control",
      "operation":"create-table",
      "partition-key-type":"task-id",
      "schema-name":"sbtest",
      "table-name":"sbtest_t1"
   }
}