DataWorks:Vetorização de dados em tarefas de sincronização offline

Etapa 1: Criar um nó de sincronização offline

1. Acesse a página Workspaces no console do DataWorks. Na barra de navegação superior, selecione a região desejada. Localize o workspace alvo e escolha Shortcuts > Data Studio na coluna Actions.

2. No diretório do projeto, clique em e escolha Create Node > Data Integration > Batch Synchronization. Configure a Source and Destination (neste tutorial, a origem é MaxCompute e o destino é Milvus), insira um Name para o nó e clique em OK.

Etapa 2: Configurar a tarefa de sincronização offline

1. Configure as informações básicas.

   • Data Source: Selecione as fontes de dados de origem e destino.

   • Running Resources: Selecione o grupo de recursos para executar a tarefa de sincronização offline. O grupo de recursos deve estar vinculado ao workspace atual e conectado à fonte de dados.

   Se não houver fontes de dados ou grupos de recursos disponíveis, certifique-se de ter concluído os pré-requisitos .

2. Configure a Data Source.

   Os parâmetros a seguir aplicam-se à fonte de dados MaxCompute usada neste tutorial. A configuração varia para outras fontes de dados; portanto, ajuste suas definições adequadamente.

   Neste tutorial, para Table, selecione test_tb. Em Filtering Method, escolha Partition Filter e defina a condição de partição como split=dev. Para When partition does not exist, selecione Ignore and run task as normal.

   Parâmetro	Descrição
   Tunnel Resource Group	Grupo de recursos para a cota do túnel. O valor padrão é Public Transmission Resource, que corresponde à cota gratuita do MaxCompute. Para mais informações sobre como selecionar um recurso de transmissão de dados para o MaxCompute, consulte Comprar e usar um grupo de recursos exclusivo do Data Transmission Service. Importante Se uma Tunnel Quota dedicada ficar indisponível devido a expiração ou pagamento em atraso, os jobs em execução alternarão automaticamente para o Public Transmission Resource.
   Tables	Selecione a tabela de origem a ser sincronizada. Se não houver tabelas de origem disponíveis, verifique se você concluiu a Preparação de dados de teste.
   Filtering Method	Suporta Partition Filter e Data Filtering: Para tabelas particionadas, selecione um intervalo de dados para sincronizar especificando partições. Para tabelas não particionadas, utilize uma cláusula WHERE para filtrar os dados a serem sincronizados.

   Clique em Data Preview para verificar se a configuração está correta.

   A caixa de diálogo Data Preview exibe os dados da tabela de origem. Neste tutorial, os dados incluem as colunas sentence (texto da avaliação), label (rótulo de sentimento) e dataset (origem do conjunto de dados).

3. Configure o Data Processing.

   1. Ative o processamento de dados. Na Data Processing List, clique em Add Node > > Data Embedding para adicionar um nó de processamento Data Embedding.

   2. Configure o nó Data Embedding. Para Name, insira sentence2emb. Em Model Name, selecione text-embedding-v4. Para o campo a ser vetorizado, escolha sentence. No Vectorization Output Field, insira sentence_e. Defina Vector Dimension como 128 e marque a caixa de seleção Convert NULL To Empty String. A tabela a seguir descreve os principais parâmetros.

      Nota

      • O desempenho do nó Data Embedding depende do desempenho do modelo configurado. Os modelos QWen fornecidos pelo Alibaba Cloud Model Studio possuem limites de consultas por segundo (QPS). Para o marketplace de modelos do Alibaba Cloud PAI, implante um modelo no PAI-EAS; seu desempenho dependerá das especificações de recursos usadas na implantação.

      • Para um determinado conjunto de parâmetros, os modelos de embedding produzem vetores determinísticos. A Integração de Dados utiliza um cache LFU (Least Frequently Used) durante a sincronização para evitar chamadas redundantes ao modelo para dados de origem idênticos. Isso melhora o desempenho do processamento e reduz os custos de embedding.

      Parâmetro	Descrição
      Model Provider	Provedor do modelo grande. Provedores suportados: Serviço de modelos do Alibaba Cloud DataWorks, Alibaba Cloud Model Studio e Marketplace de modelos do Alibaba Cloud PAI.
      Model Name	Nome do modelo de embedding. Selecione um modelo conforme suas necessidades.
      Model API Key	Chave de API usada para acessar o modelo. Obtenha a chave junto ao provedor do modelo. Alibaba Cloud Model Studio: Obter uma chave de API do Model Studio. Marketplace de modelos do Alibaba Cloud PAI: Acesse a tarefa EAS implantada, entre na depuração online e obtenha o valor do parâmetro Authorization nos cabeçalhos. Use esse valor como chave de API.
      Model endpoint	Se você selecionar Alibaba Cloud PAI model marketplace como Model Provider, configure o endpoint do modelo (endereço da API do endpoint).
      Batch Size	Tamanho do lote para vetorização. Depende do suporte do modelo de embedding a processamento em lote. O processamento em lote pode melhorar o desempenho da geração de embeddings e reduzir custos. O valor padrão é 10.
      Select Fields to Vectorize	Define quais colunas serão vetorizadas e especifica o nome do campo de saída. A Integração de Dados suporta a vetorização de um único campo de origem ou uma combinação de vários campos concatenados.
      Vectorization Output Field	Nome do campo vetorial definido para o campo de origem vetorizado.
      Vector Dimension	Dimensão do vetor de saída. O modelo de embedding deve suportar essa dimensão. O valor padrão é 1024.
      Convert a NULL value to an empty string.	Modelos grandes não processam valores NULL para vetorização. Ative esta opção para converter NULLs nos dados de origem em strings vazias e evitar erros. Esta opção vem desativada por padrão.
      Concatenate Field Names	Especifica se o nome do campo deve ser concatenado ao texto para vetorização. Se você ativar esta opção, também configure um Field Name Delimiter. Esta opção vem desativada por padrão.
      Skip Empty Fields	Ao concatenar vários campos para vetorização, define se os campos vazios devem ser ignorados. Esta opção vem ativada por padrão.

   3. Visualize a saída dos dados.

      No canto superior direito da área de configuração do nó Data Embedding, clique em Preview Data Output e depois em Preview para visualizar os resultados vetorizados e validar a configuração.

      Você também pode clicar em Dry Run na parte superior da página do editor de sincronização offline para visualizar os resultados vetorizados.

      Nos resultados da pré-visualização, a nova coluna sentence_e exibe o embedding vetorial (uma sequência de números de ponto flutuante) para cada frase.

4. Configure o Destination.

   Os parâmetros a seguir aplicam-se ao destino Milvus usado neste tutorial. A configuração varia para outros destinos; portanto, ajuste suas definições adequadamente.

   Neste tutorial, defina Collection como Milvus_Collection, selecione insert para Write Mode, configure Rows per write como 1024 e escolha Ignore para Collection creation mode.

   Parâmetro	Descrição
   Collection	Coleção que receberá os dados vetoriais.
   Partition Key	Opcional. Se a coleção for particionada, especifique uma partição para os dados vetoriais recebidos.
   Write Mode	upsert: Se o auto-ID estiver desativado para a coleção: Atualiza uma entidade com base em sua chave primária. Se o auto-ID estiver ativado para a coleção: Insere os dados como uma nova entidade com chave primária gerada automaticamente. insert: Geralmente usado para inserir dados em uma coleção com auto-ID ativado, onde o Milvus gera chaves primárias automaticamente. Usar insert em uma coleção sem auto-ID ativado causa duplicação de dados.

5. Configure o Destination Field Mapping.

   Após configurar a origem, o processamento de dados e o destino, a tarefa de sincronização offline gera automaticamente os mapeamentos de campos. Como o destino possui um esquema não estruturado, a tarefa mapeia os campos por posição por padrão. Clique em Edit ao lado de um Source Field ou Target Field para ajustar a ordem do mapeamento ou remover campos desnecessários e garantir que o mapeamento esteja correto.

   Por exemplo, neste tutorial, exclua manualmente os campos desnecessários. O mapeamento ajustado fica da seguinte forma.

   O campo de origem sentence é mapeado para o campo de destino sentence (Tipo: VarChar, maxLength: 32), e o campo de origem sentence_e é mapeado para o campo de destino sentence_e (Tipo: FloatVector, dimension: 128).

6. Configure opções adicionais em Advanced Settings.

   Clique em Advanced Settings no lado direito da página de configuração do nó. Configure parâmetros como concorrência da tarefa, taxa de sincronização e políticas de dados incorretos conforme necessário.

Etapa 3: Depurar e executar

1. No lado direito da página do editor de nós de sincronização offline, clique em Run Configuration. Defina o Resource Group e os Script Parameters para a execução de depuração e, em seguida, clique em Running na barra de ferramentas superior para testar o pipeline de sincronização.

2. Acesse o Milvus e verifique se os dados na coleção de destino estão conforme o esperado.

Etapa 4: Configurar agendamento e publicar

Clique em Scheduling Settings à direita da tarefa de sincronização offline. Defina os parâmetros da configuração de agendamento para execuções programadas, clique em Publish na barra de ferramentas superior e siga as instruções na tela no painel Publish para publicar.

Etapa 1: Criar um nó de sincronização offline

1. Acesse a página Workspaces no console do DataWorks. Na barra de navegação superior, selecione a região desejada. Localize o workspace alvo e escolha Shortcuts > Data Studio na coluna Actions.

2. No diretório do projeto, clique em e escolha Create Node > Data Integration > Batch Synchronization. Configure a Source and Destination (neste tutorial, a origem é MaxCompute e o destino é Milvus), insira um Name para o nó e clique em OK.

Configurar o script de sincronização

1. Clique em na barra de ferramentas do nó de sincronização offline para alternar para o modo script.

2. Esta tarefa sincroniza dados do MaxCompute para o Milvus.

   O script a seguir é usado neste exemplo. Para detalhes sobre o formato do script, consulte Apêndice 1: Especificação de formato do modo 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"
               }
           ]
       }
   }

   • Para detalhes sobre os parâmetros do Reader e Writer, consulte Fonte de Dados MaxCompute e Fonte de Dados Milvus.

   • Se você utilizar outros tipos de origens ou destinos, consulte a Lista de Fontes de Dados Suportadas.

   • A tabela a seguir descreve os parâmetros para o script de vetorização de dados:

     Parâmetro	Descrição	Obrigatório
     modelProvider	Especifica o provedor do modelo grande. Provedores suportados: dataworksModelService: Serviço de modelo implantado através do Serviço de Modelo de Linguagem Grande do DataWorks. bailian: Alibaba Cloud Model Studio, que suporta modelos QWen. paiModelGallery: Marketplace de Modelos do Alibaba Cloud PAI, que suporta modelos BGE-M3.	Sim
     modelName	Nome do modelo de embedding a ser utilizado. Quando modelProvider for bailian, selecione text-embedding-v4 ou text-embedding-v3. Quando modelProvider for paiModelGallery, selecione bge-m3.	Sim
     apiKey	Chave de API usada para acessar o modelo. Obtenha a chave junto ao provedor do modelo.	Sim
     endpoint	Endpoint para acessar o modelo. Obrigatório quando modelProvider estiver definido como paiModelGallery.	Não
     batchSize	Tamanho do lote para vetorização. Depende do suporte do modelo de embedding a processamento em lote. O processamento em lote pode melhorar o desempenho da geração de embeddings e reduzir custos. O valor padrão é 10.	Não
     embeddingColumns	Define as colunas a serem vetorizadas e o nome da coluna de saída. A integração de dados suporta a vetorização de uma única coluna de origem ou a concatenação de várias colunas. Exemplo: { "embeddingColumns": { "sourceColumnNames": [ "col1", "col2" ], "embeddingColumnName": "my_vector" } }	Sim
     appendDelimiter	Delimitador usado para unir valores de várias colunas em uma única string de texto para vetorização. Padrão: \n.	Não
     skipEmptyValue	Define se colunas vazias devem ser ignoradas ao concatenar várias colunas para vetorização. Padrão: false.	Não
     dimension	Dimensão do vetor de saída. O modelo de embedding deve suportar essa dimensão. O valor padrão é 1024.	Não
     nullAsEmptyString	Modelos de embedding não processam entradas NULL. Se seus dados de origem contiverem valores NULL, defina este parâmetro como true para convertê-los em strings vazias e evitar erros de vetorização. Padrão: false.	Não
     appendFieldNameEnable	Define se o nome da coluna deve ser prefixado aos dados antes da vetorização. Se for true, configure também appendFieldNameDelimiter. Padrão: false.	Não
     appendFieldNameDelimiter	Delimitador para concatenar nomes de colunas. Este parâmetro só tem efeito quando appendFieldNameEnable estiver definido como true.	Não

3. Na barra de ferramentas do nó, clique em Dry Run, depois clique em Start Collection e Preview para verificar os resultados vetorizados e validar sua configuração. Na caixa de diálogo, defina Number of samples como 3 e clique em Start Sampling para buscar os dados de entrada. A tabela de dados de entrada contém as colunas sentence, label e dataset. Os resultados da pré-visualização mostram uma nova coluna sentence_e, que exibe o vetor de ponto flutuante gerado para cada entrada de texto após a conversão de embedding. Nenhum registro de dado incorreto foi encontrado nos dados de entrada. Se os resultados estiverem corretos, clique em OK.

4. Configure opções adicionais em Advanced Settings.

   Clique em Advanced Settings no lado direito da página de configuração do nó. Configure parâmetros como concorrência da tarefa, taxa de sincronização e políticas de dados incorretos conforme necessário.

Etapa 3: Depurar e executar

1. No lado direito da página do editor de nós de sincronização offline, clique em Run Configuration. Defina o Resource Group e os Script Parameters para a execução de depuração e, em seguida, clique em Running na barra de ferramentas superior para testar o pipeline de sincronização.

2. Acesse o Milvus e verifique se os dados na coleção de destino estão conforme o esperado.

Etapa 4: Configurar agendamento e publicar

Clique em Scheduling Settings à direita da tarefa de sincronização offline. Defina os parâmetros da configuração de agendamento para execuções programadas, clique em Publish na barra de ferramentas superior e siga as instruções na tela no painel Publish para publicar.

Apêndice 1: Formato do modo script

A estrutura básica de um script é a seguinte:

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

O array steps define cada etapa de processamento. Ele deve conter pelo menos um Reader e um Writer, podendo incluir opcionalmente várias etapas Transformer intermediárias. Por exemplo, se você definir a concorrência como 2, o job terá dois fluxos paralelos de processamento de dados. Cada Reader, Transformer e Writer é considerado uma step na configuração da tarefa.

O array steps define o tipo e os parâmetros de cada etapa. O fluxo de processamento de dados segue a ordem dos elementos step no array.

Para configurações detalhadas de parâmetros dos conectores de leitura/gravação das diversas fontes de dados suportadas pela integração de dados, consulte: Fontes de Dados e Soluções de Sincronização Suportadas.

Apêndice 2: Sincronização do OSS para o Milvus

Este exemplo sincroniza dados JSON Lines (JSONL) do OSS, analisa-os, vetoriza campos especificados e grava a saída no Milvus. O exemplo a seguir mostra a configuração JSON completa:

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