DataWorks:Fonte de dados MongoDB

Tipos de dados do MongoDB compatíveis com o MongoDB Reader

O Data Integration suporta a maioria dos tipos de dados do MongoDB, mas não todos. Verifique se seus tipos de dados são compatíveis.

Ao ler tipos de dados suportados, o Data Integration 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 MongoDB. O Data Integration também converte automaticamente o tipo de dados. Não é necessário especifique a propriedade type para a coluna.

  Tipo	Leitura em lote (MongoDB Reader)	Descrição
  ObjectId	Compatível	Tipo de ID de objeto.
  Double	Compatível	Tipo de número de ponto flutuante de 64 bits.
  Inteiro de 32 bits	Compatível	Inteiro de 32 bits.
  Inteiro de 64 bits	Compatível	Inteiro de 64 bits.
  Decimal128	Compatível	Tipo Decimal128. Nota Se um campo for configurado como tipo aninhado ou combinado, ele será processado como objeto durante a serialização JSON. Adicione o parâmetro decimal128OutputType e defina-o como bigDecimal para gerar os dados como decimal.
  String	Compatível	Tipo string.
  Boolean	Compatível	Tipo booleano.
  Timestamp	Compatível	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	Compatível	Tipo data.

• Para alguns tipos de dados complexos, configure a propriedade type da coluna para realizar processamento personalizado.

  Tipo	Leitura em lote (MongoDB Reader)	Descrição
  Document	Compatível	Tipo de documento incorporado. Se a propriedade type não estiver configurada, o Document será convertido 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 visualize um exemplo detalhado, consulte Exemplo 2: Análise recursiva de Document aninhado em vários níveis abaixo.
  Array	Compatível	Tipo array. Se type estiver definido como array.json ou arrays, os dados serão processados 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 esse tipo e oferece um recurso avançado de indexação. Configure parâmetros específicos para converter strings em arrays do MongoDB. Após a conversão, grave 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: Uso do tipo combine

O tipo de dados combine do plugin MongoDB Reader permite mesclar vários campos de 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 fornece 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: Análise recursiva de Document aninhado em vários níveis

Se um documento no MongoDB tiver vários níveis de aninhamento, configure o tipo document para analisá-lo recursivamente. O código a seguir fornece 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 concluir a configuração, os campos de origem e destino são mapeados da seguinte forma: _id é mapeado para id, name é mapeado para name e a.b.c é mapeado para c.

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

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

• Para obter mais informações sobre o procedimento, consulte Configurar uma tarefa na interface sem código e Configurar 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 MongoDB.

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

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

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

É possível configurar tarefas para sincronização em lote, sincronização em tempo real completa e incremental ou sincronização em tempo real 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 Configurar uma tarefa de sincronização em tempo real para um banco de dados inteiro.

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

Para configurar uma tarefa de sincronização em lote usando o editor de código, configure os parâmetros relacionados no script com base nos requisitos unificados de formato de script. Para obter mais informações, consulte Configuração no modo de script. As informações a seguir descrevem os parâmetros que devem ser configurados para fontes de dados ao usar o editor de código para uma tarefa de sincronização em lote.

Exemplo de script do Reader

O script a seguir é um exemplo de job configurado para extrair dados do MongoDB para um ambiente local. Para obter mais informações sobre os parâmetros, consulte as descrições 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 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 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"

Ao executar um job do Data Integration, o preSql configurado é executado primeiro. A fase real de gravação de dados começa somente após a conclusão da execução do preSql. O parâmetro preSql não afeta o conteúdo dos dados gravados. O parâmetro preSql fornece execução idempotente para o Data Integration. Por exemplo, seu preSql pode limpar dados históricos antes de cada execução da tarefa com base nas suas regras de negócios. Nesse caso, se uma tarefa falhar, basta reexecutar o job do Data Integration.

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

• Configure o campo type para especifique o tipo de pré-operação. Os valores suportados são drop e remove. Exemplo: "preSql":{"type":"remove"}.

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

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

  • json: Use um objeto JSON para especifique 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]. Também é possível usar outros operadores condicionais suportados pelo 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 exclua dados com base em condições, recomendamos o uso da configuração JSON.

  • item: Configure o nome da coluna (name), a condição (condition) e o 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 executa a exclusão usando uma API padrão do MongoDB. Por exemplo: col.deleteMany(query);.

• Se o preSql não for reconhecido, nenhuma operação de pré-exclusão será realizada.