DataWorks:Source de données MongoDB

Types de données MongoDB pris en charge par MongoDB Reader

Data Integration prend en charge la plupart, mais pas la totalité, des types de données MongoDB. Assurez-vous que vos types de données sont pris en charge.

Lorsque Data Integration lit les types de données pris en charge, il effectue les opérations suivantes :

• Pour les types de données primitifs, Data Integration lit automatiquement les données à partir du chemin correspondant en fonction du nom du champ configuré dans le paramètre column. Pour plus d'informations, consultez Annexe : Exemple de script et description des paramètres pour MongoDB. Data Integration convertit également automatiquement le type de données. Vous n'avez pas besoin de spécifier la propriété type pour la colonne.

  Type	Lecture par lots (MongoDB Reader)	Description
  ObjectId	Pris en charge	Type d'identifiant d'objet.
  Double	Pris en charge	Type de nombre à virgule flottante 64 bits.
  Entier 32 bits	Pris en charge	Entier 32 bits.
  Entier 64 bits	Pris en charge	Entier 64 bits.
  Decimal128	Pris en charge	Type Decimal128. Remarque Si un champ est configuré comme un type imbriqué ou un type combiné, il est traité comme un objet lors de la sérialisation JSON. Ajoutez le paramètre decimal128OutputType et définissez-le sur bigDecimal pour exporter les données sous forme décimale.
  String	Pris en charge	Type de chaîne.
  Boolean	Pris en charge	Type booléen.
  Timestamp	Pris en charge	Type d'horodatage. Remarque BsonTimestamp stocke les horodatages. Vous n'avez pas besoin de prendre en compte l'impact des fuseaux horaires. Pour plus d'informations, consultez Problèmes de fuseau horaire dans MongoDB.
  Date	Pris en charge	Type de date.

• Pour certains types de données complexes, configurez la propriété type de la colonne afin d'effectuer un traitement personnalisé.

  Type	Lecture par lots (MongoDB Reader)	Description
  Document	Pris en charge	Type de document intégré. Si la propriété type n'est pas configurée, le Document est directement converti à l'aide de la sérialisation JSON. Si la propriété type est définie sur document, le champ est de type imbriqué. MongoDB Reader lit les propriétés du Document en fonction du chemin. Pour un exemple détaillé, consultez l'Exemple 2 : Analyse récursive d'un Document imbriqué à plusieurs niveaux ci-dessous.
  Array	Pris en charge	Type de tableau. Si le type est défini sur array.json ou arrays, les données sont directement traitées à l'aide de la sérialisation JSON. Si le type est défini sur array ou document.array, les éléments sont concaténés en une chaîne. Le séparateur, spécifié dans la propriété splitter de la colonne, est une virgule (,) par défaut.

  Important

  Data Integration ne prend pas en charge le type array (tableau). Cependant, MongoDB prend en charge ce type et propose une fonctionnalité d'indexation puissante. Configurez des paramètres spécifiques pour convertir des chaînes en tableaux MongoDB. Après la conversion, vous pouvez écrire les données dans MongoDB en parallèle.

Type de données spécial Data Integration : combine

Mappages de types de données MongoDB Reader

Le tableau suivant répertorie les mappages entre les types de données MongoDB et les types de données Data Integration pour MongoDB Reader.

Mappages de types de données MongoDB Writer

Exemple 1 : Utilisation du type combine

Le type de données combine du plug-in MongoDB Reader vous permet de fusionner plusieurs champs d'un document MongoDB en une seule chaîne JSON. Par exemple, supposons que vous souhaitiez importer des champs de trois documents MongoDB vers MaxCompute. Dans l'exemple suivant, les champs sont représentés par des clés plutôt que par des paires clé-valeur. Les champs a et b sont communs aux trois documents, et x_n est un champ variable.

• doc1: a b x_1 x_2

• doc2: a b x_2 x_3 x_4

• doc3: a b x_5

Dans le fichier de configuration, spécifiez explicitement les champs nécessitant un mappage un-à-un. Pour les champs que vous souhaitez fusionner, attribuez un nouveau nom différent de tout nom de champ existant dans le document et définissez le type sur COMBINE. Le code suivant fournit un exemple.

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

Le tableau suivant montre la sortie finale dans MaxCompute.

Exemple 2 : Analyse récursive d'un Document imbriqué à plusieurs niveaux

Si un document dans MongoDB comporte plusieurs niveaux d'imbrication, configurez le type document pour l'analyser de manière récursive. Le code suivant fournit un exemple.

• Données source dans MongoDB :

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

• Configuration de la colonne MongoDB :

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

  Une fois la configuration terminée, les champs source et destination sont mappés comme suit : _id est mappé vers id, name est mappé vers name et a.b.c est mappé vers c.

Avec la configuration précédente, la valeur du champ source imbriqué a.b.c est écrite dans le champ de destination c. Une fois la tâche de synchronisation exécutée, les données écrites dans la destination sont this is value.

Configurer une tâche de synchronisation par lots pour une table unique

• Pour plus d'informations sur la procédure, consultez Configurer une tâche dans l'interface sans code et Configurer une tâche dans l'éditeur de code.

• Pour plus d'informations sur tous les paramètres et un exemple de script pour l'éditeur de code, consultez Annexe : Exemple de script et description des paramètres pour MongoDB.

Configurer une tâche de synchronisation en temps réel pour une table unique

Consultez Configurer une tâche de synchronisation en temps réel pour une table unique.

Configurer une tâche de synchronisation pour une base de données entière

Vous pouvez configurer des tâches pour la synchronisation par lots, la synchronisation en temps réel complète et incrémentielle, ou la synchronisation en temps réel à partir de bases de données fragmentées pour une base de données entière. Pour plus d'informations, consultez Tâche de synchronisation par lots pour une base de données entière et Configurer une tâche de synchronisation en temps réel pour une base de données entière.

Configurer une tâche de synchronisation par lots à l'aide de l'éditeur de code

Si vous souhaitez configurer une tâche de synchronisation par lots à l'aide de l'éditeur de code, configurez les paramètres associés dans le script selon les exigences de format de script unifié. Pour plus d'informations, consultez Configuration en mode script. Les informations suivantes décrivent les paramètres que vous devez configurer pour les sources de données lors de la configuration d'une tâche de synchronisation par lots à l'aide de l'éditeur de code.

Exemple de script Reader

Le script suivant est un exemple d'une tâche configurée pour extraire des données de MongoDB vers un environnement local. Pour plus d'informations sur les paramètres, consultez les descriptions des paramètres qui suivent.

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

Exemple de script Writer

Le script suivant est un exemple d'une tâche de synchronisation des données configurée pour écrire des données dans MongoDB. Pour plus d'informations sur les paramètres, consultez les descriptions des paramètres qui suivent.

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

Paramètres du script Writer

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

Lors de l'exécution d'une tâche Data Integration, le preSql configuré est exécuté en premier. La phase réelle d'écriture des données ne commence qu'une fois l'exécution du preSql terminée. Le paramètre preSql n'affecte pas le contenu des données écrites. Le paramètre preSql assure l'exécution idempotente pour Data Integration. Par exemple, votre preSql peut servir à effacer les données historiques avant chaque exécution de tâche selon vos règles métier. Dans ce cas, si une tâche échoue, il vous suffit de réexécuter la tâche Data Integration.

Les exigences de format pour preSql sont les suivantes :

• Configurez le champ type pour spécifier le type d'opération préalable. Les valeurs prises en charge sont drop et remove. Exemple : "preSql":{"type":"remove"}.

  • drop : Supprime la collection et les données qu'elle contient. La collection à supprimer est spécifiée par le paramètre collectionName.

  • remove : Supprime les données en fonction d'une condition.

  • json : Utilisez un objet JSON pour spécifier les conditions de suppression des données. Exemple : "preSql":{"type":"remove", "json":"{'operationTime':{'$gte':ISODate('${last_day}T00:00:00.424+0800')}}"}. Dans cet exemple, ${last_day} est un paramètre de planification DataWorks au format $[yyyy-mm-dd]. Vous pouvez également utiliser d'autres opérateurs conditionnels pris en charge par MongoDB (tels que $gt, $lt, $gte et $lte), des opérateurs logiques (tels que and et or) ou des fonctions (telles que max, min, sum, avg et ISODate) selon vos besoins.

    Data Integration exécute la suppression des données à l'aide de l'API MongoDB standard suivante :

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

    Remarque

    Pour supprimer des données en fonction de conditions, nous vous recommandons d'utiliser la configuration JSON.

  • item : Configurez le nom de la colonne (name), la condition (condition) et la valeur de la colonne (value) pour le filtrage des données dans un élément item. Exemple : "preSql":{"type":"remove","item":[{"name":"pv","value":"100","condition":"$gt"},{"name":"pid","value":"10"}]}.

    Data Integration construit une condition de requête basée sur les conditions d'élément configurées, puis exécute la suppression à l'aide d'une API MongoDB standard. Par exemple : col.deleteMany(query);.

• Si le preSql n'est pas reconnu, aucune opération de suppression préalable n'est effectuée.