DataWorks:Fonte de dados do MongoDB

Tipos de dados do MongoDB suportados pelo MongoDB Reader

O Data Integration suporta a maioria, mas não todos, os tipos de dados do MongoDB. Certifique-se de que os seus tipos de dados tenham suporte.

Quando o Data Integration lê os tipos de dados compatíveis, ele executa as seguintes operações:

• Para tipos de dados primitivos, o Data Integration lê automaticamente os dados do caminho correspondente com base no nome do campo configurado no parâmetro column. Para obter mais informações, consulte Apêndice: Script de exemplo e descrição de parâmetros para o MongoDB. O Data Integration também converte automaticamente o tipo de dados. Não é necessário especificar a propriedade type para a coluna.

  Tipo	Leitura em lote (MongoDB Reader)	Descrição
  ObjectId	Suportado	O tipo de ID de objeto.
  Double	Suportado	O tipo de número de ponto flutuante de 64 bits.
  Inteiro de 32 bits	Suportado	Um inteiro de 32 bits.
  Inteiro de 64 bits	Suportado	Um inteiro de 64 bits.
  Decimal128	Suportado	O tipo Decimal128. Nota Se um campo for configurado como um tipo aninhado ou um tipo combine, o sistema o processa como um objeto durante a serialização JSON. Você deve adicionar o parâmetro decimal128OutputType e defini-lo como bigDecimal para exportar os dados como um decimal.
  String	Suportado	O tipo string.
  Boolean	Suportado	O tipo booleano.
  Timestamp	Suportado	O tipo timestamp. Nota O BsonTimestamp armazena timestamps. Não é necessário considerar o impacto dos fusos horários. Para obter mais informações, consulte Problemas de fuso horário no MongoDB.
  Date	Suportado	O tipo data.

• Para alguns tipos de dados complexos, você pode configurar a propriedade type para a coluna a fim de executar um processamento personalizado.

  Tipo	Leitura em lote (MongoDB Reader)	Descrição
  Document	Suportado	O tipo de documento embutido. Se a propriedade type não estiver configurada, o sistema converte o Document diretamente por serialização JSON. Se a propriedade type estiver definida como document, o campo será um tipo aninhado. O MongoDB Reader lê as propriedades do Document com base no caminho. Para ver um exemplo detalhado, consulte Exemplo 2: Analisando recursivamente um Document aninhado em vários níveis abaixo.
  Array	Suportado	O tipo array. Se type estiver definido como array.json ou arrays, o sistema processa os dados diretamente por serialização JSON. Se type estiver definido como array ou document.array, os elementos serão concatenados em uma string. O separador, especificado na propriedade splitter da coluna, é uma vírgula (,) por padrão.

  Importante

  O Data Integration não suporta o tipo array. No entanto, o MongoDB suporta o tipo array e oferece um recurso de indexação poderoso. Você pode configurar parâmetros específicos para converter strings em arrays do MongoDB. Após a conversão, você pode gravar os dados no MongoDB em paralelo.

Tipo de dados especial do Data Integration: combine

Mapeamentos de tipos de dados do MongoDB Reader

A tabela a seguir lista os mapeamentos entre os tipos de dados do MongoDB e os tipos de dados do Data Integration para o MongoDB Reader.

Mapeamentos de tipos de dados do MongoDB Writer

Exemplo 1: Usando o tipo combine

O tipo de dados combine do plugin MongoDB Reader permite mesclar vários campos em um documento do MongoDB em uma única string JSON. Por exemplo, suponha que você queira importar campos de três documentos do MongoDB para o MaxCompute. No exemplo a seguir, os campos são representados por chaves em vez de pares chave-valor. Os campos a e b são comuns a todos os três documentos, e x_n é um campo variável.

• doc1: a b x_1 x_2

• doc2: a b x_2 x_3 x_4

• doc3: a b x_5

No arquivo de configuração, especifique explicitamente os campos que exigem mapeamento um para um. Para os campos que deseja mesclar, atribua um novo nome diferente de qualquer nome de campo existente no documento e defina o tipo como COMBINE. O código a seguir apresenta um exemplo.

"column": [
{
"name": "a",
"type": "string",
},
{
"name": "b",
"type": "string",
},
{
"name": "doc",
"type": "combine",
}
]

A tabela a seguir mostra a saída final no MaxCompute.

Exemplo 2: Analisando recursivamente um Document aninhado em vários níveis

Se um documento no MongoDB tiver vários níveis de aninhamento, você pode configurar o tipo document para analisá-lo recursivamente. O código a seguir apresenta um exemplo.

• Dados de origem no MongoDB:

  {
      "name": "name1",
      "a":
      {
          "b":
          {
              "c": "this is value"
          }
      }
  }

• Configuração de coluna do MongoDB:

  {"name":"_id","type":"string"}
  {"name":"name","type":"string"}
  {"name":"a.b.c","type":"document"}

  Após a conclusão da configuração, o mapeamento dos campos de origem e de destino ocorre da seguinte forma: _id é mapeado para id, name é mapeado para name e a.b.c é mapeado para c.

Com a configuração anterior, o sistema grava o valor do campo de origem aninhado a.b.c no campo de destino c. Após a execução da tarefa de sincronização, os dados gravados no destino correspondem a this is value.

Configure uma tarefa de sincronização em lote para uma única tabela

• Para obter mais informações sobre o procedimento, consulte Configure uma tarefa na UI sem código e Configure uma tarefa no editor de código.

• Para obter mais informações sobre todos os parâmetros e um script de exemplo para o editor de código, consulte Apêndice: Script de exemplo e descrição de parâmetros para o MongoDB.

Configure uma tarefa de sincronização em tempo real para uma única tabela

Consulte Configure uma tarefa de sincronização em tempo real para uma única tabela.

Configure uma tarefa de sincronização para um banco de dados inteiro

Você pode configurar tarefas para sincronização em lote, sincronização em tempo real completa e incremental ou sincronização em tempo real a partir de bancos de dados fragmentados para um banco de dados inteiro. Para obter mais informações, consulte Tarefa de sincronização em lote para um banco de dados inteiro e Configure uma tarefa de sincronização em tempo real para um banco de dados inteiro.

Configure uma tarefa de sincronização em lote usando o editor de código

Se você quiser configurar uma tarefa de sincronização em lote usando o editor de código, deverá configurar os parâmetros relacionados no script com base nos requisitos unificados de formato de script. Para obter mais informações, consulte Configuração do modo de script. As informações a seguir descrevem os parâmetros que você deve configurar para fontes de dados ao configurar uma tarefa de sincronização em lote usando o editor de código.

Exemplo de script do Reader

O script a seguir é um exemplo de um job configurado para extrair dados do MongoDB para um ambiente local. Para obter mais informações sobre os parâmetros, consulte as descrições de parâmetros a seguir.

{
    "type":"job",
    "version":"2.0",// Version number.
    "steps":[
        {
            "category": "reader",
            "name": "Reader",
            "parameter": {
                "datasource": "datasourceName", // Data source name.
                "collectionName": "tag_data", // Collection name.
                "query": "", // Data filtering query.
                "column": [
                    {
                        "name": "unique_id", // Field name.
                        "type": "string" // Field type.
                    },
                    {
                        "name": "sid",
                        "type": "string"
                    },
                    {
                        "name": "user_id",
                        "type": "string"
                    },
                    {
                        "name": "auction_id",
                        "type": "string"
                    },
                    {
                        "name": "content_type",
                        "type": "string"
                    },
                    {
                        "name": "pool_type",
                        "type": "string"
                    },
                    {
                        "name": "frontcat_id",
                        "type": "array",
                        "splitter": ""
                    },
                    {
                        "name": "categoryid",
                        "type": "array",
                        "splitter": ""
                    },
                    {
                        "name": "gmt_create",
                        "type": "string"
                    },
                    {
                        "name": "taglist",
                        "type": "array",
                        "splitter": " "
                    },
                    {
                        "name": "property",
                        "type": "string"
                    },
                    {
                        "name": "scorea",
                        "type": "int"
                    },
                    {
                        "name": "scoreb",
                        "type": "int"
                    },
                    {
                        "name": "scorec",
                        "type": "int"
                    },
                    {
                        "name": "a.b",
                        "type": "document.int"
                    },
                    {
                        "name": "a.b.c",
                        "type": "document.array",
                        "splitter": " "
                    }
                ]
            },
            "stepType": "mongodb"
        },
        { 
            "stepType":"stream",
            "parameter":{},
            "name":"Writer",
            "category":"writer"
        }
    ],
    "setting":{
        "common": { 
            "column": { 
                "timeZone": "GMT+0" // Time zone.
            } 
        },
        "errorLimit":{
            "record":"0"// Number of error records.
        },
        "speed":{
            "throttle":true,// Specifies whether to enable throttling. If you set this parameter to false, throttling is disabled and the mbps parameter does not take effect. If you set this parameter to true, throttling is enabled.
            "concurrent":1, // Number of concurrent jobs.
            "mbps":"12"// Throttling rate. 1 mbps = 1 MB/s.
        }
    },
    "order":{
        "hops":[
            {
                "from":"Reader",
                "to":"Writer"
            }
        ]
    }
}

Exemplo de script do Writer

O script a seguir é um exemplo de um job de sincronização de dados configurado para gravar dados no MongoDB. Para obter mais informações sobre os parâmetros, consulte as descrições de parâmetros a seguir.

{
    "type": "job",
    "version": "2.0",// Version number.
    "steps": [
        {
            "stepType": "stream",
            "parameter": {},
            "name": "Reader",
            "category": "reader"
        },
        {
            "stepType": "mongodb",// Plugin name.
            "parameter": {
                "datasource": "",// Data source name.
                "column": [
                    {
                        "name": "_id",// Column name.
                        "type": "ObjectId"// Data type. If replaceKey is _id, you must set type to ObjectId. If you set type to string, the replacement fails.
                    },
                    {
                        "name": "age",
                        "type": "int"
                    },
                    {
                        "name": "id",
                        "type": "long"
                    },
                    {
                        "name": "wealth",
                        "type": "double"
                    },
                    {
                        "name": "hobby",
                        "type": "array",
                        "splitter": " "
                    },
                    {
                        "name": "valid",
                        "type": "boolean"
                    },
                    {
                        "name": "date_of_join",
                        "format": "yyyy-MM-dd HH:mm:ss",
                        "type": "date"
                    }
                ],
                "writeMode": {// Write mode.
                    "isReplace": "true",
                    "replaceKey": "_id"
                },
                "collectionName": "datax_test"// Collection name.
            },
            "name": "Writer",
            "category": "writer"
        }
    ],
    "setting": {
        "errorLimit": {// Number of error records.
            "record": "0"
        },
        "speed": {
            "throttle": true,// Specifies whether to enable throttling. If you set this parameter to false, throttling is disabled and the mbps parameter does not take effect. If you set this parameter to true, throttling is enabled.
            "concurrent": 1,// Number of concurrent jobs.
            "mbps": "1"// Throttling rate. 1 mbps = 1 MB/s.
        },
       "jvmOption": "-Xms1024m -Xmx1024m"
    },
    "order": {
        "hops": [
            {
                "from": "Reader",
                "to": "Writer"
            }
        ]
    }
}

Parâmetros do script do Writer

After applying the update, the (immutable) field '_id' was found to have been altered to _id: "2"

Quando você executa um job do Data Integration, o sistema executa o preSql configurado primeiro. A fase real de gravação de dados começa apenas após a conclusão da execução do preSql. O parâmetro preSql não afeta o conteúdo dos dados que são gravados. O parâmetro preSql garante a execução idempotente para o Data Integration. Por exemplo, seu preSql pode ser usado para limpar dados históricos antes de cada execução de tarefa com base em suas regras de negócios. Nesse caso, se uma tarefa falhar, você pode simplesmente reexecutar o job do Data Integration.

Os requisitos de formato para preSql são os seguintes:

• Você deve configurar o campo type para especificar o tipo de pré-operação. Os valores compatíveis são drop e remove. Exemplo: "preSql":{"type":"remove"}.

  • drop: Exclui a coleção e os dados nela contidos. A coleção a ser excluída é especificada pelo parâmetro collectionName.

  • remove: Exclui dados com base em uma condição.

  • json: Você pode usar um objeto JSON para especificar as condições para exclusão de dados. Exemplo: "preSql":{"type":"remove", "json":"{'operationTime':{'$gte':ISODate('${last_day}T00:00:00.424+0800')}}"}. Neste exemplo, ${last_day} é um parâmetro de agendamento do DataWorks no formato $[yyyy-mm-dd]. Você também pode usar outros operadores condicionais compatíveis com o MongoDB (como $gt, $lt, $gte e $lte), operadores lógicos (como and e or) ou funções (como max, min, sum, avg e ISODate) conforme necessário.

    O Data Integration executa a exclusão de dados usando a seguinte API padrão do MongoDB:

    query=(BasicDBObject) com.mongodb.util.JSON.parse(json);        
    col.deleteMany(query);

    Nota

    Para excluir dados com base em condições, recomendamos o uso da configuração JSON.

  • item: Você pode configurar o nome da coluna (name), condição (condition) e valor da coluna (value) para filtragem de dados em um item. Exemplo: "preSql":{"type":"remove","item":[{"name":"pv","value":"100","condition":"$gt"},{"name":"pid","value":"10"}]}.

    O Data Integration constrói uma condição de consulta com base nas condições de item configuradas e, em seguida, executa a exclusão usando uma API padrão do MongoDB. Por exemplo: col.deleteMany(query);.

• Se o sistema não reconhecer o preSql, ele não realizará nenhuma operação de pré-exclusão.