Migrate JSON-formatted data from MongoDB to MaxCompute - DataWorks: Central de Documentação do Alibaba Cloud

Pré-requisitos

Antes de começar, verifique se você:

Ativou o MaxCompute e o DataWorks. Consulte Ativar o MaxCompute e o DataWorks
Criou um fluxo de negócios no DataWorks (modo básico). Consulte Criar um fluxo de negócios

Limitações

A instância do MongoDB deve ter um endpoint de Internet ativado. O grupo de recursos padrão do DataWorks se comunica com o MongoDB pela internet pública. Instâncias exclusivas de Virtual Private Cloud (VPC) não são acessíveis.

Como funciona

O Data Integration mapeia os campos JSON do MongoDB para colunas do MaxCompute usando caminhos com notação de ponto. Por exemplo, considere este documento source:

{
  "store": {
    "bicycle": {
      "color": "red",
      "price": 19.95
    }
  },
  "expensive": 10
}

O campo store.bicycle.color é mapeado para uma coluna do MaxCompute. Especifique esse caminho na configuração do trabalho de sincronização junto com uma anotação de tipo:

Caminho do campo JSON	Anotação de tipo	Observação
`store.bicycle.color`	`document.String`	Campo aninhado — use o prefixo `document.` seguido pelo tipo de dados
`expensive`	`string`	Campo de nível superior — especifique o tipo diretamente

Etapa 1: Preparar dados de teste no MongoDB

Crie um usuário de banco de dados para autenticação do DataWorks. No shell do MongoDB, execute:
```
db.createUser({user:"bookuser",pwd:"123456",roles:["user1"]})
```
Esse comando cria um usuário chamado bookuser com a senha 123456 e a função user1.

Carregue os dados de teste na coleção. Este tutorial utiliza uma instância do ApsaraDB for MongoDB com tipo de rede VPC. Antes de prosseguir, solicite um endpoint de Internet para a instância. Sem ele, o grupo de recursos padrão do DataWorks não consegue acessar a instância. Insira o seguinte documento JSON na coleção userlog do banco de dados admin:

{
    "store": {
        "book": [
            {
                "category": "reference",
                "author": "Nigel Rees",
                "title": "Sayings of the Century",
                "price": 8.95
            },
            {
                "category": "fiction",
                "author": "Evelyn Waugh",
                "title": "Sword of Honour",
                "price": 12.99
            },
            {
                "category": "fiction",
                "author": "J. R. R. Tolkien",
                "title": "The Lord of the Rings",
                "isbn": "0-395-19395-8",
                "price": 22.99
            }
        ],
        "bicycle": {
            "color": "red",
            "price": 19.95
        }
    },
    "expensive": 10
}

Verifique se os dados foram inseridos corretamente. No console DMS para MongoDB, execute a seguinte consulta no banco de dados admin:
```
db.userlog.find().limit(10)
```

Etapa 2: Criar a tabela de destino no MaxCompute

Faça login no console do DataWorks.
No fluxo de negócios criado, clique com o botão direito e escolha New Table > MaxCompute > Table.
Na página Create Table, selecione o tipo de mecanismo e insira um nome para a tabela.
Na página de edição da tabela, clique em DDL Statement.
Na caixa de diálogo DDL Mode, insira a seguinte instrução e clique em Generate Table Schema.
```
create table mqdata (mqdata string);
```
Importante
O nome da tabela na instrução DDL deve corresponder ao nome inserido na página Create Table.
Clique em Commit To Production Environment.

Etapa 3: Adicionar a source de dados do MongoDB

Adicione o MongoDB como uma source de dados no DataWorks. Consulte Configurar uma source de dados do MongoDB.

Etapa 4: Criar e executar o trabalho de sincronização

No fluxo de negócios, clique com o botão direito no workflow e escolha Create Node > Data Integration > Offline synchronization.
Na caixa de diálogo Create Node, insira um nome e clique em Confirm.
Na barra de navegação superior, clique no ícone para alternar para o modo de script.
No modo de script, clique no ícone .
Na caixa de diálogo Import Template, defina os campos de tipo de source, source de dados, tipo de destino e source de dados e clique em Confirm.

Substitua o script gerado pela configuração a seguir. Este script lê store.bicycle.color da coleção userlog e grava o valor na coluna mqdata no MaxCompute.

{
    "type": "job",
    "steps": [
        {
            "stepType": "mongodb",
            "parameter": {
                "datasource": "mongodb_userlog",
                "column": [
                    {
                        "name": "store.bicycle.color",
                        "type": "document.String"
                    }
                ],
                "collectionName": "userlog"
            },
            "name": "Reader",
            "category": "reader"
        },
        {
            "stepType": "odps",
            "parameter": {
                "partition": "",
                "isCompress": false,
                "truncate": true,
                "datasource": "odps_first",
                "column": [
                    "mqdata"
                ],
                "emptyAsNull": false,
                "table": "mqdata"
            },
            "name": "Writer",
            "category": "writer"
        }
    ],
    "version": "2.0",
    "order": {
        "hops": [
            {
                "from": "Reader",
                "to": "Writer"
            }
        ]
    },
    "setting": {
        "errorLimit": {
            "record": ""
        },
        "speed": {
            "concurrent": 2,
            "throttle": false
        }
    }
}

Principais parâmetros:

Parâmetro	Valor	Descrição
`datasource` (leitor)	`mongodb_userlog`	Nome da source de dados do MongoDB adicionada na Etapa 3
`column[].name`	`store.bicycle.color`	Caminho com notação de ponto para o campo JSON de destino
`column[].type`	`document.String`	Anotação de tipo para campos aninhados; use `string` para campos de nível superior
`collectionName`	`userlog`	Coleção do MongoDB para leitura
`datasource` (gravador)	`odps_first`	Nome da source de dados do MaxCompute
`column` (gravador)	`mqdata`	Coluna do MaxCompute para gravação
`table`	`mqdata`	Tabela de destino no MaxCompute
`speed.concurrent`	`2`	Número de threads de leitura simultâneas

Clique no ícone para executar o trabalho.
Verifique a aba Operation Log para confirmar que o trabalho foi concluído sem erros.

Verificar o resultado

No fluxo de negócios, clique com o botão direito no workflow e escolha New > MaxCompute > ODPS SQL.
Na caixa de diálogo Create Node, insira um nome para o nó e clique em Confirm.
Na página de edição do nó ODPS SQL, insira a seguinte consulta:
```
SELECT * from mqdata;
```
Clique no ícone para executar a consulta.
Consulte a aba Operation Log para visualizar os resultados. A saída deve conter o valor red, correspondente ao valor de store.bicycle.color no documento source.

Próximos passos

Para sincronizar campos JSON adicionais, adicione mais entradas ao array column no script do trabalho de sincronização.
Para executar o trabalho de sincronização conforme uma programação, configure uma dependência de agendamento no DataWorks.