As traduções são geradas por tradução automática. Em caso de conflito entre o conteúdo da tradução e da versão original em inglês, a versão em inglês prevalecerá.

Usando o Apache Kafka como alvo para AWS Database Migration Service

Você pode usar AWS DMS para migrar dados para um cluster Apache Kafka. O Apache Kafka é uma plataforma de streaming distribuída. É possível utilizar o Apache Kafka para a ingestão e o processamento de dados de streaming em tempo real.

AWS também oferece Amazon Managed Streaming para Apache Kafka (MSKAmazon) para usar como destino. AWS DMS MSKA Amazon é um serviço de streaming Apache Kafka totalmente gerenciado que simplifica a implementação e o gerenciamento de instâncias do Apache Kafka. Ele funciona com versões de código aberto do Apache Kafka, e você acessa MSK instâncias da Amazon como AWS DMS destinos, exatamente como qualquer instância do Apache Kafka. Para obter mais informações, consulte O que é a AmazonMSK? no Guia do desenvolvedor do Amazon Managed Streaming for Apache Kafka.

Um cluster do Kafka armazena fluxos de registros em categorias chamadas tópicos que são divididos em partições. As partições são sequências de registros de dados identificadas exclusivamente (mensagens) em um tópico. As partições podem ser distribuídas entre vários agentes em um cluster para permitir o processamento paralelo dos registros de um tópico. Para obter mais informações sobre tópicos e partições e sua distribuição no Apache Kafka, consulte Tópicos e logs e Distribuição.

Seu cluster Kafka pode ser uma MSK instância da Amazon, um cluster executado em uma EC2 instância da Amazon ou um cluster local. Uma MSK instância da Amazon ou um cluster em uma EC2 instância da Amazon pode estar na mesma instância VPC ou em uma diferente. Se o cluster estiver on-premises, será possível utilizar o seu próprio servidor de nomes on-premises para a instância de replicação para resolver o nome do host do cluster. Para obter informações sobre como configurar um servidor de nomes na instância de replicação, consulte Utilização do seu próprio servidor de nomes on-premises. Para obter mais informações sobre a configuração de uma rede, consulte Configurar uma rede para uma instância de replicação.

Ao usar um MSK cluster da Amazon, certifique-se de que seu grupo de segurança permita acesso a partir da sua instância de replicação. Para obter informações sobre como alterar o grupo de segurança de um MSK cluster da Amazon, consulte Alteração do grupo de segurança de um MSK cluster da Amazon.

AWS Database Migration Service publica registros em um tópico do Kafka usando. JSON Durante a conversão, AWS DMS serializa cada registro do banco de dados de origem em um par atributo-valor no formato. JSON

Para migrar os dados de uma fonte de dados compatível para um cluster de destino do Kafka, utilize o mapeamento de objetos. Com o mapeamento de objetos, você determina como estruturar os registros de dados no tópico de destino. Também é possível definir uma chave de partição para cada tabela, que será utilizada pelo Apache Kafka para agrupar os dados em suas partições.

Atualmente, AWS DMS oferece suporte a um único tópico por tarefa. Para uma única tarefa com várias tabelas, todas as mensagens vão para um único tópico. Cada mensagem inclui uma seção de metadados que identifica o esquema e a tabela de destino. AWS DMS as versões 3.4.6 e superiores oferecem suporte à replicação multitópica usando mapeamento de objetos. Para obter mais informações, consulte Replicação de multitópico utilizando o mapeamento de objetos.

Ao usar configurações da tarefa ParallelApply*, o partition-key-type padrão é a primary-key da tabela, não o schema-name.table-name.

Conectando-se ao Kafka usando o Transport Layer Security () TLS

Um cluster Kafka aceita conexões seguras usando Transport Layer Security ()TLS. ComDMS, você pode usar qualquer uma das três opções de protocolo de segurança a seguir para proteger uma conexão de endpoint Kafka.

Usando SSL criptografia com a Amazon MSK ou um cluster Kafka autogerenciado

Você pode usar a SSL criptografia para proteger uma conexão de endpoint com a Amazon MSK ou um cluster Kafka autogerenciado. Quando você usa o método de autenticação por SSL criptografia, os clientes validam a identidade de um servidor por meio do certificado do servidor. Uma conexão criptografada é feita entre o servidor e o cliente.

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

Utilizar a autenticação SSL

Você pode usar a SSL autenticação para proteger uma conexão de endpoint com a Amazon MSK ou um cluster Kafka autogerenciado.

Para habilitar a autenticação e a criptografia do cliente usando a SSL autenticação para se conectar à AmazonMSK, faça o seguinte:

Usando SASL - SSL autenticação para se conectar à Amazon MSK

O método Simple Authentication and Security Layer (SASL) usa um nome de usuário e uma senha para validar a identidade do cliente e faz uma conexão criptografada entre o servidor e o cliente.

Para usarSASL, primeiro você cria um nome de usuário e uma senha seguros ao configurar seu MSK cluster da Amazon. Para obter uma descrição de como configurar um nome de usuário e senha seguros para um MSK cluster da Amazon, consulte Configurando SASL SCRAM /autenticando um MSK cluster da Amazon no Guia do desenvolvedor do Amazon Managed Streaming for Apache Kafka.

Crie o endpoint de destino do Kafka, defina a configuração do endpoint do protocolo de segurança (SecurityProtocol) utilizando a opção sasl-ssl. Defina também as opções SaslUsername e SaslPassword. Certifique-se de que sejam consistentes com o nome de usuário e a senha seguros que você criou quando configurou seu MSK cluster da Amazon pela primeira vez, conforme mostrado no JSON exemplo a seguir.

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

Usando uma imagem anterior para visualizar os valores originais das CDC linhas do Apache Kafka como alvo

Ao gravar CDC atualizações em um destino de streaming de dados como o Kafka, você pode visualizar os valores originais de uma linha do banco de dados de origem antes de serem alterados por uma atualização. Para tornar isso possível, AWS DMS preenche uma imagem anterior dos eventos de atualização com base nos dados fornecidos pelo mecanismo do banco de dados de origem.

Diferentes mecanismos de banco de dados de origem fornecem diferentes quantidades de informações para uma imagem anterior:

Para habilitar a criação de imagem anterior para adicionar valores originais do banco de dados de origem à saída do AWS DMS , use a configuração de tarefa BeforeImageSettings ou o parâmetro add-before-image-columns. Esse parâmetro aplica uma regra de transformação de coluna.

BeforeImageSettingsadiciona um novo JSON atributo a cada operação de atualização com valores coletados do sistema de banco de dados de origem, conforme mostrado a seguir.

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

Para opções BeforeImageSettings, aplica-se o seguinte:

Usar uma regra de transformação de imagem anterior

Como alternativa às configurações de tarefa, é possível usar o parâmetro add-before-image-columns, que aplica uma regra de transformação de coluna. Com esse parâmetro, você pode ativar antes da geração de imagens durante alvos CDC de streaming de dados, como o Kafka.

Usando add-before-image-columns em uma regra de transformação, é possível aplicar um controle mais refinado dos resultados da imagem anterior. As regras de transformação permitem que você use um localizador de objetos que oferece controle sobre as tabelas selecionadas para a regra. Além disso, é possível encadear regras de transformação, o que permite que regras diferentes sejam aplicadas a tabelas diferentes. Depois, você poderá manipular as colunas produzidas usando outras regras.

Um tipo de regra transformation com o parâmetro add-before-image-columns de uma coluna deve fornecer uma seção before-image-def. Por exemplo:

    {
      "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,
      }
    }

O valor de column-prefix precede um nome de coluna e o valor padrão de column-prefix é BI_. O valor de column-suffix é anexado ao nome da coluna e o padrão é vazio. Não defina column-prefix e column-suffix como strings vazias.

Escolha um valor para column-filter. Para adicionar somente colunas que fazem parte das chaves primárias da tabela, escolha pk-only. Escolha non-lob adicionar somente colunas que não sejam do LOB tipo. Ou escolha all para adicionar qualquer coluna que tenha um valor de imagem anterior.

Exemplo de uma regra de transformação de imagem anterior

A regra de transformação no exemplo a seguir adiciona uma nova coluna chamada BI_emp_no no destino. Portanto, uma instrução como UPDATE employees SET emp_no = 3 WHERE emp_no = 1; preenche o campo BI_emp_no com 1. Quando você grava CDC atualizações nos destinos do Amazon S3, a BI_emp_no coluna possibilita saber qual linha original foi atualizada.

{
  "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"
      }
    }
  ]
}

Para obter informações sobre como usar a ação da regra add-before-image-columns, consulte Regras de transformação e ações.

Limitações ao usar o Apache Kafka como alvo para AWS Database Migration Service

Aplicam-se as seguintes limitações ao utilizar o Apache Kafka como destino:

Utilizar o mapeamento de objetos para migrar dados para um tópico do Kafka

AWS DMS usa regras de mapeamento de tabelas para mapear dados da fonte para o tópico de destino do Kafka. Para mapear dados para um tópico de destino, utilize um tipo de regra de mapeamento de tabelas chamado mapeamento de objetos. Utilize o mapeamento de objetos para definir como os registros de dados na origem são mapeados para os registros de dados publicados em um tópico do Kafka.

Os tópicos do Kafka não têm uma estrutura predefinida além de uma chave de partição.

Para criar uma regra de mapeamento de objetos, especifique rule-type como object-mapping. Essa regra especifica o tipo de mapeamento de objeto que você deseja usar.

A estrutura da regra é a seguinte:

{
    "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 atualmente suporta map-record-to-record e map-record-to-document como os únicos valores válidos para o rule-action parâmetro. Essas configurações afetam valores que não são excluídos como parte da lista de atributos exclude-columns. Os map-record-to-document valores map-record-to-record e especificam como AWS DMS manipula esses registros por padrão. Esses valores não afetam os mapeamentos de atributos de forma alguma.

Utilize o map-record-to-record ao migrar de um banco de dados relacional para um tópico do Kafka. Esse tipo de regra utiliza o valor taskResourceId.schemaName.tableName encontrado no banco de dados relacional como a chave de partição no tópico do Kafka e cria um atributo para cada coluna no banco de dados de origem.

Ao utilizar map-record-to-record, observe o seguinte:

Uma maneira de compreender o map-record-to-record é vê-lo em ação. Para este exemplo, suponha que você está começando com uma linha de tabela do banco de dados relacional com a seguinte estrutura de dados:

Para migrar essas informações de um esquema chamado Test para um tópico do Kafka, crie regras para mapear os dados para o tópico de destino. A regra a seguir ilustra o mapeamento.

{
    "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"
            }
        }
    ]
}

Com um tópico do Kafka e uma chave de partição determinados (neste caso, taskResourceId.schemaName.tableName), o seguinte ilustra o formato do registro resultante utilizando os nossos exemplos de dados no tópico de destino do Kafka:

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

Reestruturação de dados com mapeamento de atributo

É possível reestruturar os dados ao migrá-los para um tópico do Kafka utilizando um mapa de atributos. Por exemplo, você pode combinar vários campos na origem em um único campo no destino. O mapa de atributo a seguir ilustra como reestruturar os dados.

{
    "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}"
                            }
                        }
                    }
                ]
            }
        }
    ]
}

Para definir um valor constante para partition-key, especifique um valor de partition-key. Por exemplo, é possível fazer isso para forçar o armazenamento de todos os dados em uma única partição. O mapeamento a seguir ilustra esse método.

{
    "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}"
                    }
                ]
            }
        }
    ]
}

Replicação de multitópico utilizando o mapeamento de objetos

Por padrão, AWS DMS as tarefas migram todos os dados de origem para um dos seguintes tópicos do Kafka:

Com as versões 3.4.6 e posteriores do AWS DMS mecanismo, você pode usar o kafka-target-topic atributo para mapear cada tabela de origem migrada para um tópico separado. Por exemplo, as regras de mapeamento de objetos a seguir migram as tabelas de origem Customer e Address para os tópicos customer_topic e address_topic do Kafka, respectivamente. Ao mesmo tempo, AWS DMS migra todas as outras tabelas de origem, incluindo a Bills tabela no Test esquema, para o tópico especificado no endpoint de destino.

{
    "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"
            }
        }
    ]
}

Ao utilizar a replicação de multitópico do Kafka, é possível agrupar e migrar tabelas de origem para tópicos separados do Kafka utilizando uma única tarefa de replicação.

Formato de mensagem do Apache Kafka

A JSON saída é simplesmente uma lista de pares de valores-chave.

O exemplo de JSON mensagem a seguir ilustra uma mensagem de tipo de dados com todos os metadados adicionais.

{ 
   "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"
   }
}
                

O exemplo de JSON mensagem a seguir ilustra uma mensagem de tipo de controle.

{ 
   "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"
   }
}