DataWorks:Origen de datos de MongoDB

Tipos de datos de MongoDB que admite MongoDB Reader

Data Integration admite la mayoría de los tipos de datos de MongoDB, aunque no todos. Compruebe que sus tipos de datos sean compatibles.

Cuando Data Integration lee los tipos de datos admitidos, ejecuta las siguientes operaciones:

• Para los tipos de datos primitivos, Data Integration lee automáticamente los datos de la ruta correspondiente según el nombre del campo configurado en el parámetro column. Para obtener más información, consulte Appendix: Sample script and parameter description for MongoDB. También convierte el tipo de datos de manera automática, por lo que no es necesario especificar la propiedad type de la columna.

  Tipo	Lectura por lotes (MongoDB Reader)	Descripción
  ObjectId	Compatible	Tipo de identificador de objeto.
  Double	Compatible	Tipo de número de punto flotante de 64 bits.
  Entero de 32 bits	Compatible	Número entero de 32 bits.
  Entero de 64 bits	Compatible	Número entero de 64 bits.
  Decimal128	Compatible	Tipo Decimal128. Nota Si un campo se configura como tipo anidado o de combinación, se procesa como un objeto durante la serialización JSON. Debe agregar el parámetro decimal128OutputType y establecerlo en bigDecimal para generar los datos como valor decimal.
  String	Compatible	Tipo de cadena de texto.
  Boolean	Compatible	Tipo booleano.
  Timestamp	Compatible	Tipo de marca de tiempo. Nota BsonTimestamp almacena marcas de tiempo. No es necesario considerar el impacto de las zonas horarias. Para obtener más información, consulte Problemas de zona horaria en MongoDB.
  Date	Compatible	Tipo de fecha.

• Para algunos tipos de datos complejos, puede configurar la propiedad type de la columna y aplicar un procesamiento personalizado.

  Tipo	Lectura por lotes (MongoDB Reader)	Descripción
  Document	Compatible	Tipo de documento incrustado. Si no se configura la propiedad type, el documento se convierte directamente mediante serialización JSON. Si la propiedad type se establece en document, el campo es de tipo anidado. MongoDB Reader lee las propiedades del documento según la ruta. Para ver un ejemplo detallado, consulte Example 2: Recursively parsing a multi-level nested Document más adelante.
  Array	Compatible	Tipo de arreglo. Si type se establece en array.json o arrays, los datos se procesan directamente mediante serialización JSON. Si type se establece en array o document.array, los elementos se concatenan en una cadena. El separador, definido en la propiedad splitter de la columna, es una coma (,) de forma predeterminada.

  Importante

  Data Integration no admite el tipo array. Sin embargo, MongoDB sí lo admite y ofrece una función de indexación muy potente. Puede configurar ciertos parámetros para convertir cadenas en arreglos de MongoDB. Después de la conversión, podrá escribir los datos en MongoDB de forma paralela.

Tipo de datos especial de Data Integration: combine

Asignaciones de tipos de datos de MongoDB Reader

A continuación se muestra una tabla que lista las asignaciones entre los tipos de datos de MongoDB y los de Data Integration para MongoDB Reader.

Asignaciones de tipos de datos de MongoDB Writer

Ejemplo 1: Uso del tipo combine

El tipo de datos combine del plugin MongoDB Reader permite combinar varios campos de un documento de MongoDB en una sola cadena JSON. Por ejemplo, suponga que desea importar campos de tres documentos de MongoDB a MaxCompute. En el siguiente ejemplo, los campos se representan mediante claves en lugar de pares clave-valor. Los campos a y b son comunes a los tres documentos, mientras que x_n es un campo variable.

• doc1: a b x_1 x_2

• doc2: a b x_2 x_3 x_4

• doc3: a b x_5

En el archivo de configuración, debe especificar explícitamente los campos que requieren una asignación uno a uno. Para los campos que desea combinar, asigne un nombre nuevo distinto de cualquier nombre de campo existente en el documento y establezca el tipo en COMBINE. El siguiente código muestra un ejemplo.

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

La siguiente tabla muestra la salida final en MaxCompute.

Ejemplo 2: Análisis sintáctico recursivo de un documento anidado de varios niveles

Si un documento de MongoDB tiene varios niveles de anidamiento, puede configurar el tipo document para analizarlo sintácticamente de forma recursiva. El siguiente código muestra un ejemplo.

• Datos de origen en MongoDB:

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

• Configuración de columnas de MongoDB:

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

  Una vez completada la configuración, los campos de origen y destino se asignan de la siguiente manera: _id se asigna a id, name se asigna a name y a.b.c se asigna a c.

Con la configuración anterior, el valor del campo de origen anidado a.b.c se escribe en el campo de destino c. Una vez que se ejecuta la tarea de sincronización, los datos escritos en el destino son this is value.

Configurar una tarea de sincronización por lotes para una sola tabla

• Para obtener más información sobre el procedimiento, consulte Configure a task in the codeless UI y Configure a task in the code editor.

• Para conocer todos los parámetros y ver un script de ejemplo para el editor de código, consulte Appendix: Sample script and parameter description for MongoDB.

Configurar una tarea de sincronización en tiempo real para una sola tabla

Consulte Configure a real-time synchronization task for a single table.

Configurar una tarea de sincronización para una base de datos completa

Es posible configurar tareas de sincronización por lotes, sincronización en tiempo real completa e incremental, o sincronización en tiempo real desde bases de datos fragmentadas para una base de datos completa. Para obtener más información, consulte Batch synchronization task for an entire database y Configure a real-time synchronization task for an entire database.

Configurar una tarea de sincronización por lotes mediante el editor de código

Si desea configurar una tarea de sincronización por lotes mediante el editor de código, debe definir los parámetros correspondientes en el script según los requisitos unificados de formato. Para obtener más información, consulte Script mode configuration. A continuación se describen los parámetros que debe configurar para los orígenes de datos cuando utilice el editor de código.

Script de ejemplo de Reader

El siguiente script es un ejemplo de un trabajo configurado para extraer datos de MongoDB hacia un entorno local. Para obtener más información sobre los parámetros, consulte las descripciones posteriores.

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

Script de ejemplo de Writer

El siguiente script es un ejemplo de un trabajo de sincronización de datos configurado para escribir datos en MongoDB. Para obtener más información sobre los parámetros, consulte las descripciones posteriores.

{
    "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 del script de Writer

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

Al ejecutar un trabajo de Data Integration, primero se ejecuta preSql configurado. La fase de escritura de datos comienza únicamente después de que finaliza la ejecución de preSql. El parámetro preSql no afecta el contenido de los datos escritos, mientras que preSql garantiza que la ejecución de Data Integration sea idempotente. Por ejemplo, puede usar preSql para limpiar los datos históricos antes de cada ejecución de la tarea, según sus reglas de negocio. De este modo, si una tarea falla, basta con volver a ejecutar el trabajo de Data Integration.

A continuación se describen los requisitos de formato para preSql:

• Configure el campo type para especificar el tipo de operación previa. Los valores compatibles son drop y remove. Ejemplo: "preSql":{"type":"remove"}.

  • drop: elimina la colección y los datos que contiene. La colección por eliminar se especifica mediante el parámetro collectionName.

  • remove: elimina los datos según una condición.

  • json: utilice un objeto JSON para especificar las condiciones de eliminación de datos. Ejemplo: "preSql":{"type":"remove", "json":"{'operationTime':{'$gte':ISODate('${last_day}T00:00:00.424+0800')}}"}. En este ejemplo, ${last_day} es un parámetro de programación de DataWorks con el formato $[yyyy-mm-dd]. También puede usar otros operadores condicionales compatibles con MongoDB (como $gt, $lt, $gte y $lte), operadores lógicos (como and y or) o funciones (como max, min, sum, avg e ISODate) según lo requiera.

    Data Integration ejecuta la eliminación de datos mediante la siguiente API estándar de MongoDB:

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

    Nota

    Se recomienda utilizar la configuración JSON para eliminar datos según condiciones.

  • item: dentro de un elemento item, configure el nombre de la columna (name), la condición (condition) y el valor de la columna (value) para filtrar los datos. Ejemplo: "preSql":{"type":"remove","item":[{"name":"pv","value":"100","condition":"$gt"},{"name":"pid","value":"10"}]}.

    Data Integration construye una condición de consulta a partir de los elementos item configurados y luego ejecuta la eliminación mediante una API estándar de MongoDB. Por ejemplo: col.deleteMany(query);.

• Si no se reconoce preSql, no se ejecuta ninguna operación de eliminación previa.