DataWorks:Vektorisasi data dalam tugas sinkronisasi offline

Langkah 1: Buat node sinkronisasi offline

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 ikon dan pilih Create Node > Data Integration > Batch Synchronization. Konfigurasi Source and Destination (dalam tutorial ini, sumber adalah MaxCompute dan destinasi adalah Milvus), masukkan Name untuk node tersebut, lalu klik OK.

Langkah 2: Konfigurasi tugas sinkronisasi offline

1. Konfigurasi informasi dasar.

   • Data Source: Pilih sumber data untuk sumber dan destinasi.

   • Running Resources: Pilih kelompok sumber daya untuk menjalankan tugas sinkronisasi offline. Kelompok sumber daya harus dikaitkan ke ruang kerja saat ini dan terhubung ke sumber data.

   Jika tidak ada sumber data atau kelompok sumber daya yang tersedia, pastikan Anda telah menyelesaikan prasyarat.

2. Konfigurasi Data Source.

   Parameter berikut berlaku untuk sumber data MaxCompute yang digunakan dalam tutorial ini. Konfigurasi dapat berbeda untuk sumber data lain, jadi sesuaikan pengaturan Anda.

   Dalam tutorial ini, untuk Table, pilih test_tb. Untuk Filtering Method, pilih Partition Filter dan atur kondisi partisi menjadi split=dev. Untuk When partition does not exist, pilih Ignore and run task as normal.

   Parameter	Deskripsi
   Tunnel Resource Group	Ini adalah kelompok sumber daya untuk kuota tunnel. Nilai default adalah Public Transmission Resource, yaitu kuota gratis untuk MaxCompute. Untuk informasi lebih lanjut tentang memilih sumber daya transmisi data untuk MaxCompute, lihat Beli dan gunakan kelompok sumber daya Layanan Transmisi Data eksklusif. Penting Jika Kuota Tunnel khusus tidak tersedia karena kedaluwarsa atau pembayaran tertunda, pekerjaan yang sedang berjalan secara otomatis beralih ke Public Transmission Resource.
   Tables	Pilih tabel sumber untuk disinkronkan. Jika tidak ada tabel sumber yang tersedia, pastikan Anda telah menyelesaikan Persiapan data uji.
   Filtering Method	Mendukung Partition Filter dan Data Filtering: Untuk tabel partisi, Anda dapat memilih rentang data untuk disinkronkan dengan menentukan partisi. Untuk tabel non-partisi, Anda dapat menggunakan klausa WHERE untuk memfilter data yang akan disinkronkan.

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

   Kotak dialog Data Preview menampilkan data tabel sumber. Dalam tutorial ini, data mencakup kolom sentence (teks ulasan), label (label sentimen), dan dataset (sumber dataset).

3. Konfigurasikan Data Processing.

   1. Aktifkan pemrosesan data. Di Data Processing List, klik Add Node > > Data Embedding untuk menambahkan node pemrosesan Data Embedding.

   2. Konfigurasi node Data Embedding. Untuk Name, masukkan sentence2emb. Untuk Model Name, pilih text-embedding-v4. Untuk bidang yang akan divectorisasi, pilih sentence. Untuk Vectorization Output Field, masukkan sentence_e. Atur Vector Dimension ke 128 dan centang kotak pilihan Convert NULL To Empty String. Tabel berikut menjelaskan parameter utama.

      Catatan

      • Kinerja node Data Embedding bergantung pada kinerja model yang dikonfigurasi. Model QWen yang disediakan oleh Alibaba Cloud Model Studio memiliki batas queries-per-second (QPS). Untuk pasar model Alibaba Cloud PAI, Anda harus menerapkan model di PAI-EAS, dan kinerjanya bergantung pada spesifikasi sumber daya yang digunakan untuk penerapan.

      • Untuk serangkaian parameter tertentu, model embedding menghasilkan vektor deterministik. Data Integration menggunakan cache Least Frequently Used (LFU) selama sinkronisasi untuk menghindari pemanggilan model berulang untuk data sumber yang identik. Hal ini meningkatkan kinerja pemrosesan dan mengurangi biaya embedding.

      Parameter	Deskripsi
      Model Provider	Penyedia large model. Penyedia berikut didukung: Layanan model DataWorks Alibaba Cloud, Alibaba Cloud Model Studio, dan Pasar model Alibaba Cloud PAI.
      Model Name	Nama model embedding. Pilih model sesuai kebutuhan Anda.
      Model API Key	Kunci API yang digunakan untuk mengakses model. Peroleh kunci dari penyedia model. Alibaba Cloud Model Studio: Peroleh Kunci API Model Studio. Pasar model Alibaba Cloud PAI: Buka tugas EAS yang diterapkan, masuk ke debugging online, lalu ambil nilai parameter Authorization di header. Gunakan nilai ini sebagai kunci API.
      Model endpoint	Jika Anda memilih Alibaba Cloud PAI model marketplace sebagai Model Provider, Anda harus mengonfigurasi endpoint model (alamat API endpoint).
      Batch Size	Ukuran batch untuk vektorisasi. Ini bergantung pada apakah model embedding mendukung pemrosesan batch. Pemrosesan batch dapat meningkatkan kinerja embedding dan mengurangi biaya. Nilai default adalah 10.
      Select Fields to Vectorize	Menentukan kolom mana yang akan divectorisasi dan menetapkan nama bidang output. Data Integration mendukung vektorisasi satu bidang sumber atau kombinasi beberapa bidang yang digabungkan.
      Vectorization Output Field	Nama bidang vektor yang didefinisikan untuk bidang sumber yang telah divectorisasi.
      Vector Dimension	Dimensi vektor output. Model embedding harus mendukung dimensi ini. Nilai default adalah 1024.
      Convert a NULL value to an empty string.	Large model tidak dapat memproses nilai NULL untuk vektorisasi. Aktifkan opsi ini untuk mengonversi nilai NULL dalam data sumber menjadi string kosong, sehingga mencegah error. Opsi ini dinonaktifkan secara default.
      Concatenate Field Names	Menentukan apakah nama bidang akan digabungkan dengan teks untuk vektorisasi. Jika Anda mengaktifkan opsi ini, Anda juga harus 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 output data.

      Di pojok kanan atas area konfigurasi node Data Embedding, klik Preview Data Output lalu Preview untuk melihat hasil vektorisasi dan memverifikasi konfigurasi.

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

      Dalam hasil pratinjau, kolom baru sentence_e menampilkan embedding vektor (urutan bilangan titik mengambang) untuk setiap kalimat.

4. Konfigurasi Destination.

   Parameter berikut berlaku untuk destinasi Milvus yang digunakan dalam tutorial ini. Konfigurasi dapat berbeda untuk destinasi lain, jadi sesuaikan pengaturan Anda.

   Dalam tutorial ini, atur Collection ke Milvus_Collection, pilih insert untuk Write Mode, atur Rows per write ke 1024, dan pilih Ignore untuk Collection creation mode.

   Parameter	Deskripsi
   Collection	Koleksi yang menerima data vektor.
   Partition Key	Opsional. Jika koleksi dipartisi, Anda dapat menentukan partisi untuk data vektor yang diterima.
   Write Mode	upsert: Jika auto-ID dinonaktifkan untuk koleksi: Memperbarui entitas berdasarkan kunci primernya. Jika auto-ID diaktifkan untuk koleksi: Menyisipkan data sebagai entitas baru dengan kunci primer yang dihasilkan otomatis. insert: Biasanya digunakan untuk menyisipkan data ke koleksi dengan auto-ID yang diaktifkan, di mana Milvus secara otomatis menghasilkan kunci primer. Menggunakan insert pada koleksi tanpa auto-ID yang diaktifkan menyebabkan duplikasi data.

5. Konfigurasikan Destination Field Mapping.

   Setelah Anda mengonfigurasi sumber, pemrosesan data, dan destinasi, tugas sinkronisasi offline secara otomatis menghasilkan pemetaan bidang. Karena destinasi memiliki skema tak terstruktur, tugas memetakan bidang berdasarkan posisinya secara default. Klik Edit di sebelah Source Field atau Target Field untuk menyesuaikan urutan pemetaan atau menghapus bidang yang tidak diperlukan agar pemetaan benar.

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

   Bidang sumber sentence dipetakan ke bidang destinasi sentence (Tipe: VarChar, maxLength: 32), dan bidang sumber sentence_e dipetakan ke bidang destinasi sentence_e (Tipe: FloatVector, dimensi: 128).

6. Konfigurasikan opsi lainnya di Advanced Settings.

   Klik Advanced Settings di sisi kanan halaman konfigurasi node. Anda dapat mengonfigurasi parameter seperti konkurensi tugas, laju sinkronisasi, dan kebijakan data kotor sesuai kebutuhan.

Langkah 3: Debug dan jalankan

1. Di sisi kanan halaman editor node sinkronisasi offline, klik Run Configuration. Atur Resource Group dan Script Parameters untuk menjalankan debug, lalu klik Running di bilah alat atas untuk menguji pipa sinkronisasi.

2. Buka Milvus dan periksa apakah data di koleksi destinasi sesuai harapan.

Langkah 4: Konfigurasi penjadwalan dan publikasikan

Klik Scheduling Settings di sebelah kanan tugas sinkronisasi offline. Atur parameter konfigurasi penjadwalan untuk menjalankan terjadwal, klik Publish di bilah alat atas, lalu ikuti petunjuk di layar pada panel Publish untuk mempublikasikan.

Langkah 1: Buat node sinkronisasi offline

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 ikon dan pilih Create Node > Data Integration > Batch Synchronization. Konfigurasi Source and Destination (dalam tutorial ini, sumber adalah MaxCompute dan destinasi adalah Milvus), masukkan Name untuk node tersebut, lalu klik OK.

Konfigurasi skrip sinkronisasi

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

2. Tugas ini menyinkronkan data dari MaxCompute ke Milvus.

   Skrip berikut digunakan untuk contoh ini. Untuk detail format skrip, lihat Lampiran 1: Spesifikasi format mode skrip.

   {
       "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 detail parameter Reader dan Writer, lihat Sumber Data MaxCompute dan Sumber Data Milvus.

   • Jika Anda menggunakan jenis sumber atau destinasi lain, lihat Daftar Sumber Data yang Didukung.

   • Tabel berikut menjelaskan parameter untuk skrip vektorisasi data:

     Parameter	Deskripsi	Wajib
     modelProvider	Menentukan penyedia large model. Penyedia berikut didukung: dataworksModelService: Layanan model yang diterapkan melalui Layanan Large Language Model DataWorks. bailian: Alibaba Cloud Model Studio, yang mendukung model QWen. paiModelGallery: Pasar Model Alibaba Cloud PAI, yang mendukung model BGE-M3.	Ya
     modelName	Nama model embedding yang akan digunakan. 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 yang digunakan untuk mengakses model. Peroleh kunci dari penyedia model.	Ya
     endpoint	Endpoint untuk mengakses model. Ini diperlukan saat modelProvider diatur ke paiModelGallery.	Tidak
     batchSize	Ukuran batch untuk vektorisasi. Ini bergantung pada apakah model embedding mendukung pemrosesan batch. Pemrosesan batch dapat meningkatkan kinerja embedding dan mengurangi biaya. Nilai default adalah 10.	Tidak
     embeddingColumns	Menentukan kolom yang akan divectorisasi dan nama kolom output. Integrasi data mendukung vektorisasi satu kolom sumber atau penggabungan beberapa kolom. Contoh: { "embeddingColumns": { "sourceColumnNames": [ "col1", "col2" ], "embeddingColumnName": "my_vector" } }	Ya
     appendDelimiter	Delimiter yang digunakan untuk menggabungkan nilai dari beberapa kolom menjadi satu string teks untuk vektorisasi. Default: \n.	Tidak
     skipEmptyValue	Menentukan apakah kolom kosong akan dilewati saat menggabungkan beberapa kolom untuk vektorisasi. Default: false.	Tidak
     dimension	Dimensi vektor output. Model embedding harus mendukung dimensi ini. Nilai default adalah 1024.	Tidak
     nullAsEmptyString	Model embedding tidak dapat memproses input NULL. Jika data sumber Anda berisi nilai NULL, Anda dapat mengatur parameter ini ke true untuk mengonversinya menjadi string kosong dan mencegah error vektorisasi. Default: false.	Tidak
     appendFieldNameEnable	Menentukan apakah nama kolom akan ditambahkan di awal data sebelum vektorisasi. Jika true, Anda juga harus mengonfigurasi appendFieldNameDelimiter. Default: false.	Tidak
     appendFieldNameDelimiter	Delimiter untuk menggabungkan nama kolom. Parameter ini hanya berlaku saat appendFieldNameEnable diatur ke true.	Tidak

3. Di bilah alat node, klik Dry Run, lalu klik Start Collection dan Preview untuk memeriksa hasil vektorisasi dan memverifikasi konfigurasi Anda. Di kotak dialog, atur Number of samples ke 3 dan klik Start Sampling untuk mengambil data input. Tabel data input berisi kolom sentence, label, dan dataset. Hasil pratinjau menunjukkan kolom baru sentence_e, yang menampilkan vektor bilangan titik mengambang yang dihasilkan untuk setiap entri teks setelah konversi embedding. Tidak ditemukan catatan data kotor dalam data input. Jika hasilnya benar, klik OK.

4. Konfigurasi opsi tambahan di Advanced Settings.

   Klik Advanced Settings di sisi kanan halaman konfigurasi node. Anda dapat mengonfigurasi parameter seperti konkurensi tugas, laju sinkronisasi, dan kebijakan data kotor sesuai kebutuhan.

Langkah 3: Debug dan jalankan

1. Di sisi kanan halaman editor node sinkronisasi offline, klik Run Configuration. Atur Resource Group dan Script Parameters untuk menjalankan debug, lalu klik Running di bilah alat atas untuk menguji pipa sinkronisasi.

2. Buka Milvus dan periksa apakah data di koleksi destinasi sesuai harapan.

Langkah 4: Konfigurasi penjadwalan dan publikasikan

Klik Scheduling Settings di sebelah kanan tugas sinkronisasi offline. Atur parameter konfigurasi penjadwalan untuk menjalankan terjadwal, klik Publish di bilah alat atas, lalu ikuti petunjuk di layar pada panel Publish untuk mempublikasikan.

Lampiran 1: Format mode skrip

Struktur dasar skrip 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 langkah pemrosesan. Array ini harus berisi minimal satu Reader dan satu Writer, serta dapat secara opsional mencakup beberapa langkah Transformer di antaranya. Sebagai contoh, jika Anda mengatur konkurensi ke 2, pekerjaan akan memiliki dua aliran pemrosesan data paralel. Setiap Reader, Transformer, dan Writer dianggap sebagai step dalam konfigurasi tugas.

Array steps mendefinisikan tipe dan parameter untuk setiap langkah. Alur pemrosesan data mengikuti urutan elemen step dalam array.

Untuk konfigurasi parameter terperinci konektor baca/tulis berbagai sumber data yang didukung oleh integrasi data, lihat: Sumber Data dan Solusi Sinkronisasi yang Didukung.

Lampiran 2: Sinkronisasi dari OSS ke Milvus

Contoh ini menyinkronkan data JSON Lines (JSONL) dari OSS, mengurai data tersebut, melakukan vektorisasi terhadap bidang tertentu, lalu menuliskan output ke Milvus. Contoh berikut menunjukkan konfigurasi JSON lengkap:

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