DataWorks：Vektorisasi data dengan sinkronisasi batch

Langkah 1: Buat node sinkronisasi batch

1. Masuk ke halaman Workspaces di Konsol DataWorks. Pada bilah navigasi atas, pilih Wilayah yang diinginkan. Temukan ruang kerja yang dituju, lalu pilih Shortcuts > Data Studio pada kolom Actions.

2. Di direktori proyek, klik > Create Node > Data Integration > Batch Synchronization. Konfigurasikan Source and Destination (MaxCompute sebagai sumber dan Milvus sebagai tujuan dalam tutorial ini), masukkan Name node, lalu klik OK.

Langkah 2: Konfigurasi tugas sinkronisasi batch

1. Konfigurasikan informasi dasar.

   • Data Source: Pilih Sumber Data untuk sumber dan tujuan.

   • Resource Group for Running Tasks: Pilih kelompok sumber daya untuk menjalankan tugas sinkronisasi batch. Kelompok sumber daya harus sudah di-bind ke Ruang Kerja saat ini dan memiliki konektivitas jaringan ke Sumber Data.

   Jika tidak ada Sumber Data atau kelompok sumber daya yang tersedia, pastikan Anda telah menyelesaikan langkah Sebelum memulai.

2. Konfigurasikan Source.

   Berikut adalah parameter utama untuk Sumber Data MaxCompute dalam tutorial ini. Konfigurasi dapat berbeda untuk sumber data lainnya. Harap sesuaikan parameter ini berdasarkan sumber data spesifik Anda.

   

   Parameter	Deskripsi
   Tunnel resource group	Kuota Tunnel. Secara default, public network resources dipilih, yaitu kuota gratis MaxCompute. Untuk informasi lebih lanjut tentang pemilihan sumber daya transmisi data untuk MaxCompute, lihat Beli dan gunakan kelompok sumber daya eksklusif untuk Data Transmission Service. Penting Jika Kuota Tunnel khusus menjadi tidak tersedia karena tunggakan atau kedaluwarsa, tugas akan secara otomatis beralih ke public network resources.
   Table	Pilih tabel sumber untuk disinkronkan. Jika tidak ada tabel sumber yang tersedia, pastikan Anda telah menyelesaikan langkah Siapkan data uji.
   Filter method	Mendukung Partition Filter dan Data Filter: Jika tabel sumber adalah tabel partisi, Anda dapat memilih rentang data untuk disinkronkan berdasarkan partisi. Jika tabel sumber bukan tabel partisi, Anda dapat menetapkan klausa WHERE untuk memfilter rentang data yang akan disinkronkan.

   Anda dapat mengklik Data preview untuk memeriksa apakah konfigurasi sudah benar.

   

3. Konfigurasikan Data processing.

   1. Aktifkan opsi pemrosesan data. Di Data processing list, klik Add Node > Data Vectorization untuk menambahkan node pemrosesan Data Vectorization.

      

   2. Konfigurasikan node vektorisasi data.Parameter utama dijelaskan di bawah ini:

      Catatan

      • Kinerja node vektorisasi data bergantung pada kinerja model yang dikonfigurasi. Model Qwen yang disediakan oleh platform Alibaba Cloud Bailian memiliki batas QPS. Untuk Alibaba Cloud PAI Model Gallery, Anda perlu mendeploy model di PAI-EAS sendiri, dan kinerjanya bergantung pada spesifikasi sumber daya yang digunakan untuk deployment.

      • Dengan parameter tetap, hasil vektor yang dihasilkan oleh Embedding Model bersifat deterministik. Oleh karena itu, DataWorks Data Integration mengoptimalkan proses sinkronisasi dengan cache Least Frequently Used (LFU) untuk data input identik. Caching ini menghindari pemanggilan model berulang, meningkatkan kinerja pemrosesan, dan mengurangi biaya embedding.

      Parameter	Deskripsi
      Model Provider	Penyedia LLM. Penyedia berikut saat ini didukung: Alibaba Cloud DataWorks Model Service, platform Alibaba Cloud Bailian, dan Alibaba Cloud PAI Model Gallery.
      Model Name	Nama Embedding Model. Pilih sesuai kebutuhan.
      Model API Key	Kunci API untuk mengakses model. Peroleh dari penyedia model. Platform Alibaba Cloud Bailian: Peroleh Kunci API Bailian. Alibaba Cloud PAI Model Gallery: Buka tugas EAS yang telah dideploy, masuk ke halaman debugging online, lalu ambil nilai parameter Authorization dari header. Gunakan nilai ini sebagai Kunci API.
      Model Endpoint	Saat Anda memilih Alibaba Cloud PAI Model Gallery sebagai Model Provider, Anda harus mengonfigurasi titik akhir akses model.
      Batch Size	Jumlah catatan yang diproses dalam setiap batch. Efektivitas parameter ini bergantung pada apakah model dasar mendukung batching. Pemrosesan batch dapat meningkatkan kinerja embedding dan mengurangi biaya. Nilai default adalah 10.
      Select the fields to be vectorized	Tentukan kolom mana yang akan divectorisasi dan tentukan nama bidang output. Data Integration mendukung vektorisasi satu bidang atau beberapa bidang yang digabung dari sumber.
      Vectorization Result Field	Nama untuk bidang baru yang akan berisi vektor output.
      Vector Dimension	Dimensi vektor output. Ini harus sesuai dengan dimensi yang didukung oleh model yang dipilih. Nilai default adalah 1024.
      Convert NULL values to empty strings	LLM tidak dapat memproses nilai NULL. Jika data sumber Anda berisi NULL, aktifkan opsi ini untuk mengonversinya menjadi string kosong guna menghindari error. Opsi ini dinonaktifkan secara default.
      Concatenate field names	Menentukan apakah nama bidang akan digabungkan dengan teks untuk vektorisasi. Jika dipilih, Anda juga perlu mengonfigurasi Field name delimiter. Opsi ini dinonaktifkan secara default.
      Skip empty fields	Saat menggabungkan beberapa bidang untuk vektorisasi, menentukan apakah bidang kosong akan dilewati. Opsi ini diaktifkan secara default.

   3. Pratinjau data output.

      Klik Preview Output Data di pojok kanan atas area konfigurasi node vektorisasi data, lalu klik Preview untuk melihat pratinjau output hasil vektorisasi dan memverifikasi konfigurasi Anda.

      Anda juga dapat mengklik Dry Run di bagian atas halaman edit sinkronisasi batch untuk melihat pratinjau hasil vektorisasi.

      

4. Konfigurasikan Destination.

   Berikut adalah parameter utama untuk Sumber Data Milvus dalam tutorial ini. Konfigurasi dapat berbeda untuk sumber data lainnya. Harap sesuaikan sesuai sumber data Anda.

   

   Parameter	Deskripsi
   Collection	Koleksi untuk menerima data vektor.
   Partition Key	Opsional. Jika koleksi dipartisi, Anda dapat menentukan partisi untuk data vektor yang masuk.
   Write Mode	upsert: Jika koleksi tidak memiliki auto ID yang diaktifkan: Memperbarui entitas dalam koleksi berdasarkan kunci primer. Jika koleksi memiliki auto ID yang diaktifkan: Mengganti kunci primer dalam entitas dengan kunci yang di-generate secara otomatis dan memasukkan data. insert: Biasanya digunakan untuk memasukkan data ke koleksi dengan auto ID yang diaktifkan. Milvus secara otomatis meng-generate kunci primer. Menggunakan insert pada koleksi tanpa auto ID yang diaktifkan dapat menyebabkan data duplikat.

5. Konfigurasikan Destination field mapping.

   Setelah mengonfigurasi sumber, pemrosesan, dan tujuan, tugas sinkronisasi batch secara otomatis menghasilkan pemetaan bidang. Karena tujuan adalah sumber data tanpa skema tetap, bidang dipetakan berdasarkan posisi secara default. Klik Edit di samping Source field atau Destination field untuk menyesuaikan urutan pemetaan bidang atau menghapus bidang yang tidak diperlukan guna memastikan pemetaan sudah benar.

   Sebagai contoh, dalam tutorial ini, bidang yang tidak diperlukan dihapus secara manual. Pemetaan yang telah disesuaikan adalah sebagai berikut.

   

6. Konfigurasikan Advanced settings.

   Klik Advanced Settings di sisi kanan halaman konfigurasi node. Anda dapat mengatur parameter seperti konkurensi tugas, laju sinkronisasi, dan kebijakan Dirty Data sesuai kebutuhan.

Langkah 3: Jalankan dan debug

1. Klik Run Configuration di sisi kanan halaman edit node sinkronisasi batch. Atur Resource Group dan Script Parameters untuk menjalankan debug. Lalu, klik Run di bilah alat atas untuk menjalankan pipeline.

2. Verifikasi di Milvus bahwa data dalam koleksi tujuan sudah benar.

Langkah 4: Konfigurasi penjadwalan dan publikasi

Klik Scheduling di sisi kanan tugas sinkronisasi batch. Atur parameter Scheduling Configuration untuk menjalankan secara periodik. Lalu, klik Publish di bilah alat atas untuk membuka panel publikasi dan selesaikan proses publikasi sesuai petunjuk.

Langkah 1: Buat node sinkronisasi batch

1. Buka halaman Ruang Kerja di konsol DataWorks. Di bilah navigasi atas, pilih wilayah yang diinginkan. Temukan ruang kerja yang diinginkan dan pilih Shortcuts > Data Studio di kolom Actions.

2. Di direktori proyek, klik > Create Node > Data Integration > Batch Synchronization. Konfigurasikan Source and Destination (MaxCompute sebagai sumber dan Milvus sebagai tujuan dalam tutorial ini), masukkan Name node, lalu klik OK.

Langkah 2: Konfigurasi skrip sinkronisasi

1. Klik ikon di bilah alat node sinkronisasi batch untuk beralih ke mode pengeditan skrip.

2. Konfigurasikan tugas sinkronisasi batch dari MaxCompute ke Milvus.

   Konfigurasikan skrip JSON untuk tugas sinkronisasi batch sesuai dengan Lampiran 1: Format script mode. Skrip untuk contoh ini adalah sebagai berikut:

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

   • Untuk penjelasan parameter di bagian Reader dan Writer, lihat Sumber data MaxCompute dan Sumber data Milvus.

   • Jika Anda menggunakan jenis sumber data dan tujuan lainnya, lihat Daftar sumber data.

   • Tabel berikut menjelaskan parameter skrip untuk transformer vektorisasi data (embedding-transformer):

     Parameter	Deskripsi	Wajib
     modelProvider	Penyedia LLM. Penyedia berikut saat ini didukung: dataworksModelService: Layanan model yang dideploy melalui Layanan Large Model DataWorks. bailian: Platform Alibaba Cloud Bailian, yang mendukung model Qwen. paiModelGallery: Alibaba Cloud PAI Model Gallery, yang mendukung model BGE-M3.	Ya
     modelName	Nama Embedding Model. Saat modelProvider adalah bailian, Anda dapat memilih text-embedding-v4 atau text-embedding-v3. Saat modelProvider adalah paiModelGallery, Anda dapat memilih bge-m3.	Ya
     apiKey	Kunci API untuk mengakses model. Peroleh dari penyedia model.	Ya
     endpoint	Saat modelProvider adalah paiModelGallery, Anda perlu mengonfigurasi titik akhir akses (alamat API Endpoint) untuk model.	Tidak
     batchSize	Jumlah catatan yang diproses dalam setiap batch. Efektivitas parameter ini bergantung pada apakah model dasar mendukung batching. Pemrosesan batch dapat meningkatkan kinerja embedding dan mengurangi biaya. Nilai default adalah 10.	Tidak
     embeddingColumns	Menentukan kolom mana yang akan divectorisasi dan menentukan nama bidang output. Data Integration mendukung vektorisasi satu bidang atau beberapa bidang yang digabung dari sumber. Contoh: { "embeddingColumns": { "sourceColumnNames": [ "col1", "col2" ], "embeddingColumnName": "my_vector" } }	Ya
     appendDelimiter	Delimiter yang digunakan saat menggabungkan beberapa nilai bidang menjadi satu teks untuk vektorisasi. Nilai default adalah \n.	Tidak
     skipEmptyValue	Saat menggabungkan beberapa bidang untuk vektorisasi, menentukan apakah bidang kosong akan dilewati. Nilai default adalah false.	Tidak
     dimension	Dimensi vektor output. Ini harus sesuai dengan dimensi yang didukung oleh model yang dipilih. Nilai default adalah 1024.	Tidak
     nullAsEmptyString	LLM tidak mengizinkan nilai NULL untuk embedding. Jika data sumber berisi NULL, Anda dapat mengaktifkan opsi ini untuk mengonversinya menjadi string kosong guna menghindari error. Nilai default adalah false.	Tidak
     appendFieldNameEnable	Menentukan apakah data asli akan digabungkan dengan nama bidang untuk vektorisasi. Jika diaktifkan, Anda juga perlu mengonfigurasi appendFieldNameDelimiter. Nilai default adalah false.	Tidak
     appendFieldNameDelimiter	Delimiter yang digunakan saat menggabungkan nama bidang. Ini hanya berlaku saat appendFieldNameEnable diaktifkan.	Tidak

3. Lakukan Dry Run.

   Klik Dry Run di bagian atas halaman edit node sinkronisasi batch. Lalu, klik Start Acquisition dan Preview untuk melihat hasil vektorisasi dan memastikan konfigurasi sudah benar.

4. Konfigurasikan Advanced settings.

   Klik Advanced Settings di sisi kanan halaman konfigurasi node. Anda dapat mengatur parameter seperti konkurensi tugas, laju sinkronisasi, dan kebijakan Dirty Data sesuai kebutuhan.

Langkah 3: Jalankan dan debug

1. Klik Run Configuration di sisi kanan halaman edit node sinkronisasi batch. Atur Resource Group dan Script Parameters untuk menjalankan debug. Lalu, klik Run di bilah alat atas untuk menjalankan pipeline.

2. Verifikasi di Milvus bahwa data dalam koleksi tujuan sudah benar.

Langkah 4: Konfigurasi penjadwalan dan publikasi

Klik Scheduling di sisi kanan tugas sinkronisasi batch. Atur parameter Scheduling Configuration untuk menjalankan secara periodik. Lalu, klik Publish di bilah alat atas untuk membuka panel publikasi dan selesaikan proses publikasi sesuai petunjuk.

Lampiran 1: Format script mode

Struktur dasar Script Mode adalah sebagai berikut:

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

Array steps mendefinisikan setiap operator pemrosesan dalam pipeline. Array ini harus berisi minimal satu Reader dan satu Writer, dengan sejumlah operator Transformer di antaranya. Jika Anda mengatur konkurensi menjadi 2, job akan memiliki dua aliran pemrosesan data paralel. Setiap operator (Reader, Transformer, atau Writer) didefinisikan sebagai objek step.

steps mendefinisikan tipe dan parameter setiap operator. Sinkronisasi dan pemrosesan data mengikuti urutan setiap step dalam konfigurasi JSON secara ketat.

Untuk konfigurasi parameter detail berbagai channel reader dan writer Sumber Data yang didukung oleh Data Integration, lihat: Sumber data yang didukung dan solusi sinkronisasi.

Lampiran 2: Contoh OSS ke Milvus

Contoh ini menunjukkan cara menyinkronkan data JSONL dari Object Storage Service (OSS) ke Milvus. Pipeline ini mengurai JSON, melakukan vektorisasi bidang tertentu, dan menulis data akhir ke Milvus. Contoh konfigurasi JSON lengkap adalah sebagai berikut:

{
    "type": "job",
    "version": "2.0",
    "steps": [
        {
            "stepType": "oss",
            "parameter": {
                "datasource": "${Your_OSS_Data_Source_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_Data_Source_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
        }
    }
}