Copiar dados do MongoDB, usando o Azure Data Factory ou Synapse Analytics

APLICA-SE A: Azure Data Factory Azure Synapse Analytics

Dica

Experimente o Data Factory no Microsoft Fabric, uma solução de análise tudo-em-um para empresas. O Microsoft Fabric abrange desde movimentação de dados até ciência de dados, análise em tempo real, business intelligence e relatórios. Saiba como iniciar uma avaliação gratuita!

Este artigo descreve como usar a atividade de cópia nos pipelines do Azure Data Factory e do Azure Synapse Analytics para copiar dados de um banco de dados do MongoDB. Ele amplia o artigo Visão geral da atividade de cópia que apresenta uma visão geral da atividade de cópia.

Importante

O novo conector do MongoDB fornece suporte ao MongoDB nativo aprimorado. Se você estiver usando o conector MongoDB herdado em sua solução, que é tem suporte no estado em que se encontra para oferecer compatibilidade com versões anteriores, confira o artigo Conector do MongoDB (herdado).

Funcionalidades com suporte

O conector MongoDB é compatível com as seguintes funcionalidades:

Funcionalidades com suporte IR
Atividade de cópia (origem/coletor) ① ②

① Runtime de integração do Azure ② Runtime de integração auto-hospedada

Para obter uma lista de armazenamentos de dados com suporte como origens/coletores, consulte a tabela de Armazenamentos de dados com suporte.

Especificamente, este conector do MongoDB dá suporte a versões até 4.2. Se o seu trabalho exige versões mais recentes que a 4.2, considere utilizar o MongoDB Atlas com o conector do MongoDB Atlas, que oferece suporte e recursos mais completos.

Pré-requisitos

Se o armazenamento de dados estiver localizado dentro de uma rede local, em uma rede virtual do Azure ou na Amazon Virtual Private Cloud, você precisará configurar um runtime de integração auto-hospedada para se conectar a ele.

Se o armazenamento de dados for um serviço de dados de nuvem gerenciado, você poderá usar o Azure Integration Runtime. Se o acesso for restrito aos IPs que estão aprovados nas regras de firewall, você poderá adicionar IPs do Azure Integration Runtime à lista de permissões.

Você também pode usar o recurso de runtime de integração da rede virtual gerenciada no Azure Data Factory para acessar a rede local sem instalar e configurar um runtime de integração auto-hospedada.

Para obter mais informações sobre os mecanismos de segurança de rede e as opções compatíveis com o Data Factory, consulte Estratégias de acesso a dados.

Introdução

Para executar a atividade de Cópia com um pipeline, será possível usar as ferramentas ou os SDKs abaixo:

Criar um serviço vinculado ao MongoDB usando a interface do usuário

Use as etapas a seguir para criar um serviço vinculado ao MongoDB na interface do usuário portal do Microsoft Azure.

  1. Navegue até a guia Gerenciar em seu espaço de trabalho do Azure Data Factory ou do Synapse e selecione Serviços Vinculados, em seguida, clique em Novo:

  2. Pesquise por MongoDB e selecione o conector MongoDB.

    Selecione o conector MongoDB.

  3. Configure os detalhes do serviço, teste a conexão e crie o novo serviço vinculado.

    Configure um serviço vinculado ao MongoDB.

Detalhes da configuração do conector

As seções a seguir fornecem detalhes sobre as propriedades usadas para definir entidades do Data Factory específicas ao conector do MongoDB.

Propriedades do serviço vinculado

As propriedades a seguir têm suporte para o serviço vinculado do MongoDB:

Propriedade Descrição Obrigatório
type A propriedade type deve ser definida como: MongoDbV2 Sim
connectionString Especifique a cadeia de conexão MongoDB, por exemplo, mongodb://[username:password@]host[:port][/[database][?options]]. Confira o manual do MongoDB na cadeia de conexão para obter mais detalhes.

Você também pode colocar uma cadeia de conexão no Azure Key Vault. Consulte o artigo Credenciais de armazenamento no Azure Key Vault para saber mais detalhes.
Sim
Banco de Dados O nome do banco de dados que você deseja criar. Sim
connectVia O Integration Runtime a ser usado para se conectar ao armazenamento de dados. Saiba mais na seção Pré-requisitos. Se não for especificado, ele usa o Integration Runtime padrão do Azure. Não

Exemplo:

{
    "name": "MongoDBLinkedService",
    "properties": {
        "type": "MongoDbV2",
        "typeProperties": {
            "connectionString": "mongodb://[username:password@]host[:port][/[database][?options]]",
            "database": "myDatabase"
        },
        "connectVia": {
            "referenceName": "<name of Integration Runtime>",
            "type": "IntegrationRuntimeReference"
        }
    }
}

Propriedades do conjunto de dados

Para obter uma lista completa de seções e propriedades disponíveis para definição de conjuntos de dados, consulte Conjuntos de dados e serviços vinculados. As propriedades a seguir têm suporte para o conjunto de dados do MongoDB:

Propriedade Descrição Obrigatório
type A propriedade type do conjunto de dados deve ser definida como: MongoDbV2Collection Sim
collectionName Nome da coleção no banco de dados MongoDB. Sim

Exemplo:

{
    "name": "MongoDbDataset",
    "properties": {
        "type": "MongoDbV2Collection",
        "typeProperties": {
            "collectionName": "<Collection name>"
        },
        "schema": [],
        "linkedServiceName": {
            "referenceName": "<MongoDB linked service name>",
            "type": "LinkedServiceReference"
        }
    }
}

Propriedades da atividade de cópia

Para obter uma lista completa das seções e propriedades disponíveis para definir atividades, confia o artigo Pipelines. Esta seção fornece uma lista das propriedades com suporte pela fonte e pelo coletor do MongoDB.

MongoDB como fonte

As propriedades a seguir têm suporte na seção source da atividade de cópia:

Propriedade Descrição Obrigatório
type A propriedade type da fonte da atividade de cópia deve ser definida como: MongoDbV2Source Sim
filter Especifica o filtro de seleção usando operadores de consulta. Para retornar todos os documentos em uma coleção, omita esse parâmetro ou passe um documento vazio ({}). Não
cursorMethods.project Especifica os campos a serem retornados nos documentos para projeção. Para retornar todos os campos nos documentos correspondentes, omita este parâmetro. Não
cursorMethods.sort Especifica a ordem na qual a consulta retorna documentos correspondentes. Consulte cursor.sort(). Não
cursorMethods.limit Especifica o número máximo de documentos que o servidor retorna. Consulte cursor.limit(). Não
cursorMethods.skip Especifica o número de documentos a serem ignorados e de onde o MongoDB começa a retornar resultados. Consulte cursor.skip(). Não
batchSize Especifica o número de documentos a serem retornados em cada lote da resposta da instância do MongoDB. Na maioria dos casos, modificar o tamanho do lote não afetará o usuário ou o aplicativo. O Cosmos DB limita que cada lote não possa exceder 40 MB de tamanho, que é a soma do tamanho do batchSize dos documentos; portanto, diminua esse valor se o tamanho do documento for grande. Não
(o padrão é 100)

Dica

Suporte do ADF consumindo o documento BSON em Modo estrito. Verifique se sua consulta de filtro está em Modo estrito em vez do modo Shell. Veja mais descrições no manual do MongoDB.

Exemplo:

"activities":[
    {
        "name": "CopyFromMongoDB",
        "type": "Copy",
        "inputs": [
            {
                "referenceName": "<MongoDB input dataset name>",
                "type": "DatasetReference"
            }
        ],
        "outputs": [
            {
                "referenceName": "<output dataset name>",
                "type": "DatasetReference"
            }
        ],
        "typeProperties": {
            "source": {
                "type": "MongoDbV2Source",
                "filter": "{datetimeData: {$gte: ISODate(\"2018-12-11T00:00:00.000Z\"),$lt: ISODate(\"2018-12-12T00:00:00.000Z\")}, _id: ObjectId(\"5acd7c3d0000000000000000\") }",
                "cursorMethods": {
                    "project": "{ _id : 1, name : 1, age: 1, datetimeData: 1 }",
                    "sort": "{ age : 1 }",
                    "skip": 3,
                    "limit": 3
                }
            },
            "sink": {
                "type": "<sink type>"
            }
        }
    }
]

MongoDB como coletor

As seguintes propriedades são suportadas na seção Copy Activity sink:

Propriedade Descrição Obrigatório
type O tipo de propriedade do coletor da atividade de cópia deve ser definido como MongoDbV2Sink. Sim
writeBehavior Descreve como gravar dados no MongoDB. Valores permitidos são insert e upsert.

O comportamento de upsert será substituir o documento se um documento com a mesma _id já existir; caso contrário, insira o documento.

Nota: o serviço gera automaticamente uma _id ID para um documento _id se não for especificada uma ID no documento original ou no mapeamento de coluna. Isso significa que, para upsert funcionar conforme esperado, o documento deve ter uma ID.
No
(o padrão é insert)
writeBatchSize A propriedade writeBatchSize controla o tamanho dos documentos que escrevemos em cada lote. Você pode tentar aumentar o valor de writeBatchSize para melhorar o desempenho e diminuir o valor se o tamanho do documento for grande. Não
(o padrão é 10.000)
writeBatchTimeout O tempo de espera para que a operação de inserção em lote seja concluída antes de atingir o tempo limite. O valor permitido é timespan. Não
(o padrão é 30:00:00 - 30 minutos)

Dica

Para importar documentos JSON como eles estão, consulte a seção importar ou exportar documentos JSON; para copiar de dados em formato tabular, consulte mapeamento de esquema.

Exemplo

"activities":[
    {
        "name": "CopyToMongoDB",
        "type": "Copy",
        "inputs": [
            {
                "referenceName": "<input dataset name>",
                "type": "DatasetReference"
            }
        ],
        "outputs": [
            {
                "referenceName": "<Document DB output dataset name>",
                "type": "DatasetReference"
            }
        ],
        "typeProperties": {
            "source": {
                "type": "<source type>"
            },
            "sink": {
                "type": "MongoDbV2Sink",
                "writeBehavior": "upsert"
            }
        }
    }
]

Importar e exportar documentos JSON

Você pode usar esse conector do MongoDB para facilmente:

  • Copiar documentos entre duas coleções do MongoDB no estado em que se encontram.
  • Importar documentos JSON de várias fontes para o MongoDB, incluindo o Azure Cosmos DB, o Armazenamento de blobs do Azure, o Azure Data Lake Storage e outros repositórios baseados em arquivo compatíveis.
  • Exportar documentos JSON de uma coleção do MongoDB para vários repositórios baseados em arquivo.

Para efetuar essa cópia independente de esquema, ignore a seção da "estrutura" (também chamada de esquema) no conjunto de dados e mapeamento de esquema na atividade de cópia.

Mapeamento de esquema

Para copiar dados do MongoDB para o coletor de tabela ou revertido, confira o mapeamento do esquema.

Atualizar o serviço vinculado do MongoDB

Aqui estão as etapas que ajudarão você a atualizar seu serviço vinculado e as consultas relacionadas:

  1. Crie um novo serviço vinculado ao MongoDB e configure-o consultando as Propriedades do serviço vinculado.

  2. Se utilizar consultas no SQL em seus pipelines que se referem ao antigo serviço vinculado do MongoDB, substitua-as pelas consultas equivalentes do MongoDB. Confira na tabela a seguir os exemplos de substituição:

    consulta SQL Consulta equivalente do MongoDB
    SELECT * FROM users db.users.find({})
    SELECT username, age FROM users db.users.find({}, {username: 1, age: 1})
    SELECT username AS User, age AS Age, statusNumber AS Status, CASE WHEN Status = 0 THEN "Pending" CASE WHEN Status = 1 THEN "Finished" ELSE "Unknown" END AS statusEnum LastUpdatedTime + interval '2' hour AS NewLastUpdatedTime FROM users db.users.aggregate([{ $project: { _id: 0, User: "$username", Age: "$age", Status: "$statusNumber", statusEnum: { $switch: { branches: [ { case: { $eq: ["$Status", 0] }, then: "Pending" }, { case: { $eq: ["$Status", 1] }, then: "Finished" } ], default: "Unknown" } }, NewLastUpdatedTime: { $add: ["$LastUpdatedTime", 2 * 60 * 60 * 1000] } } }])
    SELECT employees.name, departments.name AS department_name FROM employees LEFT JOIN departments ON employees.department_id = departments.id; db.employees.aggregate([ { $lookup: { from: "departments", localField: "department_id", foreignField: "_id", as: "department" } }, { $unwind: "$department" }, { $project: { _id: 0, name: 1, department_name: "$department.name" } } ])

Para obter uma lista de armazenamentos de dados com suporte como coletores e fontes da atividade de cópia, confira os armazenamentos de dados com suporte.