DataWorks:Vectorizing data in offline synchronization tasks

Étape 1 : Créer un nœud de synchronisation hors ligne

1. Accédez à la page Workspaces dans la console DataWorks. Dans la barre de navigation supérieure, sélectionnez la région souhaitée. Localisez l'espace de travail souhaité et choisissez Shortcuts > Data Studio dans la colonne Actions.

2. Dans le répertoire du projet, cliquez sur l'icône et choisissez Create Node > Data Integration > Batch Synchronization. Configurez la Source and Destination (dans ce tutoriel, la source est MaxCompute et la destination est Milvus), saisissez un Name pour le nœud, puis cliquez sur OK.

Étape 2 : Configurer la tâche de synchronisation hors ligne

1. Configurez les informations de base.

   • Data Source : sélectionnez les sources de données pour la source et la destination.

   • Running Resources : sélectionnez le groupe de ressources pour exécuter la tâche de synchronisation hors ligne. Le groupe de ressources doit être lié à l'espace de travail actuel et connecté à la source de données.

   Si aucune source de données ou aucun groupe de ressources n'est disponible, assurez-vous d'avoir terminé les prérequis .

2. Configurez la Data Source.

   Les paramètres suivants s'appliquent à la source de données MaxCompute utilisée dans ce tutoriel. La configuration varie pour les autres sources de données, adaptez donc vos paramètres en conséquence.

   Dans ce tutoriel, pour Table, sélectionnez test_tb. Pour Filtering Method, sélectionnez Partition Filter et définissez la condition de partition sur split=dev. Pour When partition does not exist, sélectionnez Ignore and run task as normal.

   Paramètre	Description
   Tunnel Resource Group	Il s'agit du groupe de ressources pour le quota de tunnel. La valeur par défaut est Public Transmission Resource, qui correspond au quota gratuit pour MaxCompute. Pour plus d'informations sur la sélection d'une ressource de transmission de données pour MaxCompute, consultez Acheter et utiliser un groupe de ressources Data Transmission Service exclusif. Important Si un quota de tunnel dédié devient indisponible en raison d'une expiration ou d'un impayé, les jobs en cours basculent automatiquement vers la ressource Public Transmission Resource.
   Tables	Sélectionnez la table source à synchroniser. Si aucune table source n'est disponible, assurez-vous d'avoir terminé la section Préparation des données de test.
   Filtering Method	Prend en charge Partition Filter et Data Filtering : Pour les tables partitionnées, vous pouvez sélectionner une plage de données à synchroniser en spécifiant les partitions. Pour les tables non partitionnées, vous pouvez utiliser une clause WHERE pour filtrer les données à synchroniser.

   Vous pouvez cliquer sur Data Preview pour vérifier si la configuration est correcte.

   La boîte de dialogue Data Preview affiche les données de la table source. Dans ce tutoriel, les données incluent les colonnes sentence (texte de l'avis), label (étiquette de sentiment) et dataset (source du jeu de données).

3. Configurez le Data Processing.

   1. Activez le traitement des données. Dans la Data Processing List, cliquez sur Add Node > > Data Embedding pour ajouter un nœud de traitement Data Embedding.

   2. Configurez le nœud Data Embedding. Pour Name, saisissez sentence2emb. Pour Model Name, sélectionnez text-embedding-v4. Pour le champ à vectoriser, sélectionnez sentence. Pour Vectorization Output Field, saisissez sentence_e. Définissez Vector Dimension sur 128 et cochez la case Convert NULL To Empty String. Le tableau suivant décrit les paramètres clés.

      Remarque

      • Les performances du nœud Data Embedding dépendent des performances du modèle configuré. Les modèles QWen fournis par Alibaba Cloud Model Studio ont des limites de requêtes par seconde (QPS). Pour la marketplace de modèles Alibaba Cloud PAI, vous devez déployer un modèle sur PAI-EAS, et ses performances dépendent des spécifications de ressources utilisées pour le déploiement.

      • Pour un ensemble de paramètres donné, les modèles d'embedding produisent des vecteurs déterministes. Data Integration utilise un cache Least Frequently Used (LFU) pendant la synchronisation pour éviter les appels de modèle redondants pour des données source identiques. Cela améliore les performances de traitement et réduit les coûts d'embedding.

      Paramètre	Description
      Model Provider	Le fournisseur de grand modèle. Les fournisseurs suivants sont pris en charge : Service de modèle DataWorks d'Alibaba Cloud, Alibaba Cloud Model Studio et Marketplace de modèles Alibaba Cloud PAI.
      Model Name	Le nom du modèle d'embedding. Sélectionnez un modèle en fonction de vos besoins.
      Model API Key	La clé API utilisée pour accéder au modèle. Obtenez la clé auprès du fournisseur de modèle. Alibaba Cloud Model Studio : Obtenir une clé API Model Studio. Marketplace de modèles Alibaba Cloud PAI : accédez à la tâche EAS déployée, entrez dans le débogage en ligne et obtenez la valeur du paramètre Authorization dans les en-têtes. Utilisez cette valeur comme clé API.
      Model endpoint	Si vous sélectionnez Alibaba Cloud PAI model marketplace comme Model Provider, vous devez configurer le point de terminaison du modèle (l'adresse API du point de terminaison).
      Batch Size	La taille du lot pour la vectorisation. Cela dépend de la prise en charge du traitement par lots par le modèle d'embedding. Le traitement par lots peut améliorer les performances d'embedding et réduire les coûts. La valeur par défaut est 10.
      Select Fields to Vectorize	Définit les colonnes à vectoriser et spécifie le nom du champ de sortie. Data Integration prend en charge la vectorisation d'un seul champ source ou d'une combinaison de plusieurs champs concaténés.
      Vectorization Output Field	Le nom du champ vectoriel défini pour le champ source vectorisé.
      Vector Dimension	La dimension du vecteur de sortie. Le modèle d'embedding doit prendre en charge cette dimension. La valeur par défaut est 1024.
      Convert a NULL value to an empty string.	Les grands modèles ne peuvent pas traiter les valeurs NULL pour la vectorisation. Activez cette option pour convertir les valeurs NULL de vos données source en chaînes vides, ce qui évite les erreurs. Cette option est désactivée par défaut.
      Concatenate Field Names	Spécifie s'il faut concaténer le nom du champ avec le texte pour la vectorisation. Si vous activez cette option, vous devez également configurer un Field Name Delimiter. Cette option est désactivée par défaut.
      Skip Empty Fields	Lors de la concaténation de plusieurs champs pour la vectorisation, spécifie s'il faut ignorer les champs vides. Cette option est activée par défaut.

   3. Prévisualisez la sortie des données.

      Dans le coin supérieur droit de la zone de configuration du nœud Data Embedding, cliquez sur Preview Data Output puis sur Preview pour afficher les résultats vectorisés et vérifier la configuration.

      Vous pouvez également cliquer sur Dry Run en haut de la page de l'éditeur de synchronisation hors ligne pour prévisualiser les résultats vectorisés.

      Dans les résultats de prévisualisation, la nouvelle colonne sentence_e affiche l'embedding vectoriel (une séquence de nombres à virgule flottante) pour chaque phrase.

4. Configurez la Destination.

   Les paramètres suivants s'appliquent à la destination Milvus utilisée dans ce tutoriel. La configuration varie pour les autres destinations, adaptez donc vos paramètres en conséquence.

   Dans ce tutoriel, définissez Collection sur Milvus_Collection, sélectionnez insert pour Write Mode, définissez Rows per write sur 1024 et sélectionnez Ignore pour Collection creation mode.

   Paramètre	Description
   Collection	La collection qui reçoit les données vectorielles.
   Partition Key	Facultatif. Si la collection est partitionnée, vous pouvez spécifier une partition pour les données vectorielles reçues.
   Write Mode	upsert : Si l'ID automatique est désactivé pour la collection : met à jour une entité en fonction de sa clé primaire. Si l'ID automatique est activé pour la collection : insère les données en tant que nouvelle entité avec une clé primaire générée automatiquement. insert : généralement utilisé pour insérer des données dans une collection avec l'ID automatique activé, où Milvus génère automatiquement les clés primaires. L'utilisation de insert sur une collection sans ID automatique activé entraîne une duplication des données.

5. Configurez le Destination Field Mapping.

   Après avoir configuré la source, le traitement des données et la destination, la tâche de synchronisation hors ligne génère automatiquement les mappages de champs. Étant donné que la destination possède un schéma non structuré, la tâche mappe les champs par leur position par défaut. Cliquez sur Edit à côté d'un Source Field ou d'un Target Field pour ajuster l'ordre de mappage ou supprimer les champs inutiles afin de garantir que le mappage est correct.

   Par exemple, dans ce tutoriel, vous supprimez manuellement les champs inutiles. Le mappage ajusté est le suivant.

   Le champ source sentence est mappé au champ de destination sentence (Type : VarChar, maxLength : 32), et le champ source sentence_e est mappé au champ de destination sentence_e (Type : FloatVector, dimension : 128).

6. Configurez davantage d'options dans les Advanced Settings.

   Cliquez sur Advanced Settings sur le côté droit de la page de configuration du nœud. Vous pouvez configurer des paramètres tels que la concurrence des tâches, le taux de synchronisation et les politiques de données erronées selon vos besoins.

Étape 3 : Déboguer et exécuter

1. Sur le côté droit de la page de l'éditeur du nœud de synchronisation hors ligne, cliquez sur Run Configuration. Définissez le Resource Group et les Script Parameters pour l'exécution de débogage, puis cliquez sur Running dans la barre d'outils supérieure pour tester le pipeline de synchronisation.

2. Accédez à Milvus et vérifiez si les données dans la collection de destination sont conformes aux attentes.

Étape 4 : Configurer la planification et publier

Cliquez sur Scheduling Settings à droite de la tâche de synchronisation hors ligne. Définissez les paramètres de configuration de planification pour les exécutions planifiées, cliquez sur Publish dans la barre d'outils supérieure, puis suivez les invites à l'écran dans le panneau Publish pour publier.

Étape 1 : Créer un nœud de synchronisation hors ligne

1. Accédez à la page Workspaces dans la console DataWorks. Dans la barre de navigation supérieure, sélectionnez la région souhaitée. Localisez l'espace de travail souhaité et choisissez Shortcuts > Data Studio dans la colonne Actions.

2. Dans le répertoire du projet, cliquez sur l'icône et choisissez Create Node > Data Integration > Batch Synchronization. Configurez la Source and Destination (dans ce tutoriel, la source est MaxCompute et la destination est Milvus), saisissez un Name pour le nœud, puis cliquez sur OK.

Configurer le script de synchronisation

1. Cliquez sur l'icône dans la barre d'outils du nœud de synchronisation hors ligne pour passer en mode script.

2. Cette tâche synchronise les données de MaxCompute vers Milvus.

   Le script suivant est utilisé pour cet exemple. Pour plus de détails sur le format du script, consultez la section Annexe 1 : Spécification du format en mode script.

   {
       "type": "job",
       "version": "2.0",
       "steps": [
           {
               "stepType": "odps",
               "parameter": {
                   "partition": [
                       "split=dev"
                   ],
                   "datasource": "MaxCompute_Source",
                   "successOnNoPartition": true,
                   "tunnelQuota": "default",
                   "column": [
                       "sentence"
                   ],
                   "enableWhere": false,
                   "table": "test_tb"
               },
               "name": "Reader",
               "category": "reader"
           },
           {
               "category": "flatmap",
               "stepType": "embedding-transformer",
               "parameter": {
                   "modelProvider": "bailian",
                   "modelName": "text-embedding-v4",
                   "embeddingColumns": {
                       "sourceColumnNames": [
                           "sentence"
                       ],
                       "embeddingColumnName": "sentence_e"
                   },
                   "apiKey": "sk-****",
                   "dimension": 128,
                   "nullAsEmptyString": true
               },
               "displayName": "sentence2emb",
               "description": ""
           },
           {
               "stepType": "milvus",
               "parameter": {
                   "schemaCreateMode": "ignore",
                   "enableDynamicSchema": true,
                   "datasource": "Milvus_Source",
                   "column": [
                       {
                           "name": "sentence",
                           "type": "VarChar",
                           "elementType": "None",
                           "maxLength": "32"
                       },
                       {
                           "name": "sentence_e",
                           "type": "FloatVector",
                           "dimension": "128",
                           "elementType": "None",
                           "maxLength": "65535"
                       }
                   ],
                   "writeMode": "insert",
                   "collection": "Milvus_Collection",
                   "batchSize": 1024,
                   "columnMapping": [
                       {
                           "sourceColName": "sentence",
                           "dstColName": "sentence"
                       },
                       {
                           "sourceColName": "sentence_e",
                           "dstColName": "sentence_e"
                       }
                   ]
               },
               "name": "Writer",
               "category": "writer"
           }
       ],
       "setting": {
           "errorLimit": {
               "record": "0"
           },
           "speed": {
               "concurrent": 2,
               "throttle": false
           }
       },
       "order": {
           "hops": [
               {
                   "from": "Reader",
                   "to": "Writer"
               }
           ]
       }
   }

   • Pour plus de détails sur les paramètres Reader et Writer, consultez les sections Source de données MaxCompute et Source de données Milvus.

   • Si vous utilisez d'autres types de sources ou de destinations, consultez la liste des sources de données prises en charge.

   • Le tableau suivant décrit les paramètres du script de vectorisation des données :

     Paramètre	Description	Obligatoire
     modelProvider	Spécifie le fournisseur de grand modèle. Les fournisseurs suivants sont pris en charge : dataworksModelService : un service de modèle déployé via DataWorks Large Language Model Service. bailian : Alibaba Cloud Model Studio, qui prend en charge les modèles QWen. paiModelGallery : Marketplace de modèles Alibaba Cloud PAI, qui prend en charge les modèles BGE-M3.	Oui
     modelName	Le nom du modèle d'embedding à utiliser. Lorsque modelProvider est bailian, vous pouvez sélectionner text-embedding-v4 ou text-embedding-v3. Lorsque modelProvider est paiModelGallery, vous pouvez sélectionner bge-m3.	Oui
     apiKey	La clé API utilisée pour accéder au modèle. Obtenez la clé auprès du fournisseur de modèle.	Oui
     endpoint	Le point de terminaison pour accéder au modèle. Ceci est requis lorsque modelProvider est défini sur paiModelGallery.	Non
     batchSize	La taille du lot pour la vectorisation. Cela dépend de la prise en charge du traitement par lots par le modèle d'embedding. Le traitement par lots peut améliorer les performances d'embedding et réduire les coûts. La valeur par défaut est 10.	Non
     embeddingColumns	Définit les colonnes à vectoriser et le nom de la colonne de sortie. Data Integration prend en charge la vectorisation d'une seule colonne source ou la concaténation de plusieurs colonnes. Exemple : { "embeddingColumns": { "sourceColumnNames": [ "col1", "col2" ], "embeddingColumnName": "my_vector" } }	Oui
     appendDelimiter	Le délimiteur utilisé pour joindre les valeurs de plusieurs colonnes en une seule chaîne de texte pour la vectorisation. Par défaut : \n.	Non
     skipEmptyValue	Spécifie s'il faut ignorer les colonnes vides lors de la concaténation de plusieurs colonnes pour la vectorisation. Par défaut : false.	Non
     dimension	La dimension du vecteur de sortie. Le modèle d'embedding doit prendre en charge cette dimension. La valeur par défaut est 1024.	Non
     nullAsEmptyString	Les modèles d'embedding ne peuvent pas traiter les entrées NULL. Si vos données source contiennent des valeurs NULL, vous pouvez définir ce paramètre sur true pour les convertir en chaînes vides et éviter les erreurs de vectorisation. Par défaut : false.	Non
     appendFieldNameEnable	Spécifie s'il faut préfixer le nom de la colonne aux données avant la vectorisation. Si la valeur est true, vous devez également configurer appendFieldNameDelimiter. Par défaut : false.	Non
     appendFieldNameDelimiter	Le délimiteur pour la concaténation des noms de colonnes. Ce paramètre prend effet uniquement lorsque appendFieldNameEnable est défini sur true.	Non

3. Dans la barre d'outils du nœud, cliquez sur Dry Run, puis cliquez sur Start Collection et Preview pour vérifier les résultats vectorisés et valider votre configuration. Dans la boîte de dialogue, définissez Number of samples sur 3 et cliquez sur Start Sampling pour récupérer les données d'entrée. La table de données d'entrée contient les colonnes sentence, label et dataset. Les résultats de prévisualisation affichent une nouvelle colonne sentence_e, qui présente le vecteur à virgule flottante généré pour chaque entrée de texte après la conversion en embedding. Aucune donnée erronée n'a été trouvée dans les données d'entrée. Si les résultats sont corrects, cliquez sur OK.

4. Configurez davantage d'options dans les Advanced Settings.

   Cliquez sur Advanced Settings sur le côté droit de la page de configuration du nœud. Vous pouvez configurer des paramètres tels que la concurrence des tâches, le taux de synchronisation et les politiques de données erronées selon vos besoins.

Étape 3 : Déboguer et exécuter

1. Sur le côté droit de la page de l'éditeur du nœud de synchronisation hors ligne, cliquez sur Run Configuration. Définissez le Resource Group et les Script Parameters pour l'exécution de débogage, puis cliquez sur Running dans la barre d'outils supérieure pour tester le pipeline de synchronisation.

2. Accédez à Milvus et vérifiez si les données dans la collection de destination sont conformes aux attentes.

Étape 4 : Configurer la planification et publier

Cliquez sur Scheduling Settings à droite de la tâche de synchronisation hors ligne. Définissez les paramètres de configuration de planification pour les exécutions planifiées, cliquez sur Publish dans la barre d'outils supérieure, puis suivez les invites à l'écran dans le panneau Publish pour publier.

Annexe 1 : Format en mode script

La structure de base d'un script est la suivante :

{
    "type": "job", 
    "version": "2.0", 
    "steps": [
        {
            "stepType": "xxx",
            "parameter": {
            },
            "name": "Reader",
            "category": "reader"
        },
        {
            "stepType": "xxx",
            "parameter": {
            },
            "name": "transformer1",
            "category": "map/flatmap"
        },
        {
            "stepType": "xxx",
            "parameter": {
            },
            "name": "transformer2",
            "category": "map/flatmap"
        },
        {
            "stepType": "xxx",
            "parameter": {
            },
            "name": "Writer",
            "category": "writer"
        }
    ],
    "setting": {
    }
}

Le tableau steps définit chaque étape de traitement. Il doit contenir au moins un Reader et un Writer, et peut éventuellement inclure plusieurs étapes Transformer intermédiaires. Par exemple, si vous définissez la concurrence sur 2, le job disposera de deux flux de traitement de données parallèles. Chaque Reader, Transformer et Writer est considéré comme une step dans la configuration de la tâche.

Le tableau steps définit le type et les paramètres de chaque étape. Le flux de traitement des données suit l'ordre des éléments step dans le tableau.

Pour les configurations détaillées des paramètres des connecteurs de lecture/écriture des diverses sources de données prises en charge par Data Integration, consultez : Sources de données prises en charge et solutions de synchronisation.

Annexe 2 : Synchronisation depuis OSS vers Milvus

Cet exemple synchronise des données JSON Lines (JSONL) depuis OSS, les analyse, vectorise les champs spécifiés et écrit la sortie dans Milvus. L'exemple suivant montre la configuration JSON complète :

{
    "type": "job",
    "version": "2.0",
    "steps": [
        {
            "stepType": "oss",
            "parameter": {
                "datasource": "${YOUR_OSS_DATASOURCE_NAME}",
                "column": [
                    {
                        "name": "chunk_text",
                        "index": 0,
                        "type": "string"
                    }
                ],
                "fieldDelimiter": ",",
                "encoding": "UTF-8",
                "fileFormat": "jsonl",
                "object": [
                    "embedding/chunk1.jsonl"
                ]
            },
            "name": "Reader",
            "category": "reader"
        },
        {
            "stepType": "json-extracting",
            "parameter": {
                "column": [
                    {
                        "name": "text",
                        "fromColumn": "chunk_text",
                        "jsonPath": "$.text",
                        "type": "STRING",
                        "nullOrInvalidDataAction": "DIRTY_DATA"
                    }
                ]
            },
            "name": "jsonextract",
            "category": "flatmap"
        },
        {
            "stepType": "embedding-transformer",
            "parameter": {
                "modelProvider": "bailian",
                "modelName": "text-embedding-v4",
                "apiKey": "${YOUR_API_KEY}",
                "embeddingColumns": {
                    "sourceColumnNames": [
                        "text"
                    ],
                    "embeddingColumnName": "my_vector"
                },
                "batchSize": 8,
                "dimension": 1024
            },
            "name": "embedding",
            "category": "flatmap"
        },
        {
            "stepType": "milvus",
            "parameter": {
                "schemaCreateMode": "ignore",
                "enableDynamicSchema": true,
                "datasource": "${YOUR_MILVUS_DATASOURCE_NAME}",
                "column": [
                    {
                        "name": "my_vector",
                        "type": "FloatVector",
                        "dimension": "1024",
                        "elementType": "None",
                        "maxLength": "65535"
                    },
                    {
                        "name": "text",
                        "type": "VarChar",
                        "elementType": "None",
                        "maxLength": "65535"
                    }
                ],
                "collection": "yunshi_vector_07171130",
                "writeMode": "insert",
                "batchSize": 1024,
                "columnMapping": [
                    {
                        "sourceColName": "my_vector",
                        "dstColName": "my_vector"
                    },
                    {
                        "sourceColName": "text",
                        "dstColName": "text"
                    }
                ]
            },
            "name": "Writer",
            "category": "writer"
        }
    ],
    "setting": {
        "errorLimit": {
            "record": "0"
        },
        "speed": {
            "concurrent": 1
        }
    }
}