Objek tugas DTS untuk migrasi, sinkronisasi, atau langganan - Data Transmission Service

DTS mendukung pemanggilan operasi API untuk mengonfigurasi atau mengkueri objek migrasi, sinkronisasi, atau langganan dari tugas DTS. Topik ini menjelaskan operasi API terkait serta menyediakan definisi dan contoh konfigurasi objek.

Operasi API dan parameter terkait

API	Deskripsi
Konfigurasikan tugas migrasi atau sinkronisasi DTS Konfigurasikan tugas langganan DTS	Konfigurasikan objek migrasi, sinkronisasi, atau langganan dari tugas DTS dalam parameter permintaan `Dblist`.
Kueri detail tugas DTS Kueri daftar tugas DTS dan detail eksekusi setiap tugas	Kueri objek migrasi, sinkronisasi, atau langganan dari tugas DTS dalam parameter respons `DbObject`.

Definisi objek migrasi, sinkronisasi, atau langganan

Nilai parameter yang terkait dengan objek berupa string JSON. Bagian berikut menjelaskan parameter tersebut.

Jika objek migrasi, sinkronisasi, atau langganan mencakup beberapa database, Anda dapat merujuk pada definisi berikut:

Penting

Instansi langganan tidak mendukung fitur pemetaan. Nilai parameter name untuk instansi langganan harus sama dengan nama database atau tabel yang akan dilanggan.

{
    "Nama database 1 yang akan dimigrasikan, disinkronkan, atau dilanggan": {
        "name": "Nama database 1 di instans tujuan",
        "all": true (Menunjukkan seluruh database dimigrasikan, disinkronkan, atau dilanggan)
    },
    "Nama database 2 yang akan dimigrasikan, disinkronkan, atau dilanggan": {
        "name": "Nama database 2 di instans tujuan",
        "all": false (Menunjukkan seluruh database tidak dimigrasikan, disinkronkan, atau dilanggan),
        "Table": {
            "Nama tabel A yang akan dimigrasikan, disinkronkan, atau dilanggan": {
                "name": "Nama tabel A di instans tujuan",
                "all": true (Menunjukkan seluruh tabel dimigrasikan, disinkronkan, atau dilanggan),
                "dml_op": "Operasi DML yang akan dimigrasikan atau disinkronkan secara inkremental",
                "ddl_op": "Operasi DDL yang akan dimigrasikan atau disinkronkan secara inkremental"
            }
        }
    },
    "Nama database 3 yang akan dimigrasikan, disinkronkan, atau dilanggan": {
        "name": "Nama database 3 di instans tujuan",
        "all": true (Menunjukkan seluruh database dimigrasikan, disinkronkan, atau dilanggan),
        "dml_op": "Operasi DML yang akan dimigrasikan atau disinkronkan secara inkremental",
        "ddl_op": "Operasi DDL yang akan dimigrasikan atau disinkronkan secara inkremental"
    }
}

Jika objek migrasi atau sinkronisasi berada pada level kolom atau mencakup kondisi filter, Anda dapat merujuk pada definisi berikut:

{
    "Nama database yang akan dimigrasikan, disinkronkan, atau dilanggan": {
        "name": "Nama database di instans tujuan",
        "all": false (Menunjukkan seluruh database tidak dimigrasikan, disinkronkan, atau dilanggan),
        "Table": {
            "Nama tabel A yang akan dimigrasikan, disinkronkan, atau dilanggan": {
                "name": "Nama tabel A di instans tujuan",
                "all": false (Menunjukkan seluruh tabel tidak dimigrasikan, disinkronkan, atau dilanggan),
                "filter": "id>10"
                "column": {
                    "id": {
                        "key": "PRI",
                        "name": "id",
                        "type": "int(11)",
                        "sharedKey": false,
                        "state": "checked"
                    }
                },
                "shard": 12
            }
        }
    }
}

Jika instans database tujuan dari objek migrasi atau sinkronisasi adalah AnalyticDB for MySQL atau AnalyticDB for PostgreSQL, Anda dapat merujuk pada definisi berikut:

{
    "Nama database yang akan dimigrasikan atau disinkronkan": {
        "name": "Nama database di instans tujuan",
        "all": false (Tetap sebagai false. Terlepas dari apakah objek berada pada level database atau tabel, jika instans tujuan adalah AnalyticDB for MySQL atau AnalyticDB for PostgreSQL, parameter ini tetap sebagai false, dan Anda juga harus menentukan informasi seperti kunci partisi tabel),
        "Table": {
            "Nama tabel A yang akan dimigrasikan atau disinkronkan": {
                "all": true (Menunjukkan seluruh tabel dimigrasikan atau disinkronkan),
                "name": "Nama tabel A di instans tujuan",
                "primary_key": "id (Menentukan kunci primer)",
                "type": "dimension (Jenis tabel)",
            }
            "Nama tabel B yang akan dimigrasikan atau disinkronkan": {
                "all": true (Menunjukkan seluruh tabel dimigrasikan atau disinkronkan),
                "name": "Nama tabel B di instans tujuan",
                "part_key": "id (Menentukan kunci partisi)",
                "primary_key": "id (Menentukan kunci primer)",
                "type": "partition (Jenis tabel)",
                "tagColumnValue": "Nilai kolom tag"
            }
        }
    }
}

Jika Anda perlu menetapkan kebijakan resolusi konflik independen untuk objek sinkronisasi, Anda dapat merujuk pada definisi berikut:

Catatan

Fitur ini hanya didukung untuk instansi sinkronisasi dua arah antara instansi MySQL atau antara kluster PolarDB for MySQL.
Anda dapat menetapkan kebijakan resolusi konflik independen pada level database atau tabel.
Untuk kolom yang dikonfigurasi dengan kebijakan resolusi konflik independen, kebijakan resolusi konflik global tidak berlaku.

Pengaturan level tabel

{
    "Nama database 1 yang akan disinkronkan": {
      "name": "Nama database 1 di instans tujuan",
      "all": true (Menunjukkan seluruh database disinkronkan),
      "conflict": "Kebijakan resolusi konflik level tugas"
    },
    "Nama database 2 yang akan disinkronkan": {
      "name": "Nama database 2 di instans tujuan",
      "all": false (Menunjukkan seluruh database tidak disinkronkan),
      "conflict": "overwrite",
      "Table": {
        "Nama tabel A yang akan disinkronkan": {
          "name": "Nama tabel A di instans tujuan",
          "all": true (Menunjukkan seluruh tabel disinkronkan),
          "cdr_cmp_col": "Kolom deteksi konflik",
          "cdr_rslv_col": "Kolom deteksi konflik",
          "resolve_method": "Kebijakan resolusi konflik level tabel"
        }
      }
    }
}

Pengaturan level database

Saat objek sinkronisasi adalah seluruh database:

"Nama database 1 yang akan disinkronkan": {
  "name": "Nama database 1 di instans tujuan",
  "all": true (Menunjukkan seluruh database disinkronkan),
  "conflict": "Kebijakan resolusi konflik level tugas",
  "cdr_cmp_col": "Kolom deteksi konflik",
  "cdr_rslv_col": "Kolom deteksi konflik",
  "resolve_method": "Kebijakan resolusi konflik level database"
  }
}

Saat objek sinkronisasi bukan seluruh database:

"Nama database 2 yang akan disinkronkan": {
  "name": "Nama database 2 di instans tujuan",
  "all": false (Menunjukkan seluruh database tidak disinkronkan),
  "conflict": "Kebijakan resolusi konflik level tugas",
  "cdr_cmp_col": "Kolom deteksi konflik",
  "cdr_rslv_col": "Kolom deteksi konflik",
  "resolve_method": "Kebijakan resolusi konflik level database",
  "Table": {
    "Nama tabel A yang akan disinkronkan": {
      "name": "Nama tabel A di instans tujuan",
      "all": true (Menunjukkan seluruh tabel disinkronkan)
    }
  }
}

Jika Anda perlu mengonfigurasi tugas integrasi data ke data lake, Anda dapat merujuk pada definisi berikut:

Parameter	Deskripsi
`write_operation`	Metode yang digunakan untuk menulis data saat terjadi konflik data. `append`: Mempertahankan data saat ini di database tujuan dan menambahkan data baru. `overwrite`: Menimpa data yang bertentangan di database tujuan. `errorIfExists`: Tugas melaporkan kesalahan dan keluar. `ignore`: Melewati operasi penulisan data saat ini, melanjutkan eksekusi, dan menggunakan data yang bertentangan di database tujuan.
`targetType`	Format data setelah ditulis ke OSS (konversi paksa). Format yang didukung: `Byte`, `Integer`, `Long`, `Double`, `String`, `Binary`, `Boolean`, `Timestamp`, dan `Date`. Catatan Jika parameter ini tidak ditentukan, DTS secara otomatis mengonversi tipe data dari sumber ke tipe yang didukung.
`etl_date`	Nama kolom tambahan (konstan) yang akan ditambahkan. Catatan Nilai dari dua parameter `etl_date` harus sama.
`syntacticType`	Tetap sebagai ADD, yang menunjukkan bahwa kolom ditambahkan. Catatan Nilai kolom yang ditambahkan (`value`) hanya boleh berupa konstanta dan harus diapit tanda kutip tunggal ('').
`part_key`	Kunci partisi tabel tujuan. Parameter ini memiliki dua nilai yang mungkin. Catatan Parameter ini wajib hanya ketika tabel tujuan adalah tabel partisi. Kolom yang akan diintegrasikan dari sumber. Kolom konstan yang ditambahkan ke tujuan dan ditetapkan sebagai kunci partisi. Catatan Formatnya adalah `<Key>=<Value>`. Misalnya, `dt=2025-07-07` menunjukkan kolom konstan bernama `dt` dengan nilai `2025-07-07`.

{
    "Nama database 1 yang akan diintegrasikan": {
        "all": false (Tetap sebagai false), 
        "Table": {
            "Nama tabel A yang akan diintegrasikan": {
                "all": false (Menunjukkan seluruh tabel tidak diintegrasikan), 
                "filter": "", 
                "write_operation": "Metode yang digunakan untuk menulis data", 
                "name": "Nama tabel A di instans tujuan", 
                "column": {
                    "Nama kolom a yang akan diintegrasikan": {
                        "name": "Nama kolom a di instans tujuan", 
                        "targetType": "Tipe kolom di OSS"
                    }, 
                    ******, 
                    "etl_date": {
                        "syntacticType": "ADD (Tetap sebagai ADD)", 
                        "name": "etl_date", 
                        "type": "String (Tetap sebagai String)", 
                        "value": "'2025-07-08 03:30:00'"
                    }
                }, 
                "part_key": "dt=2025-07-07"
            }
        }, 
        "name": "dtstestdata (Nama database 1 di instans tujuan)"
    }
}

Parameter	Deskripsi
`name`	Nama tempat database, tabel, atau kolom sumber dipetakan di tujuan. Misalnya, jika Anda ingin memigrasikan database bernama dtssource ke database bernama dtstarget, Anda harus mengatur parameter name menjadi dtstarget.
`all`	Menentukan apakah memilih semua tabel atau kolom. Nilai yang valid: `true`: ya. Catatan Jika Anda memilih semua tabel atau kolom, Anda tidak perlu mengonfigurasi informasi tabel atau kolom tertentu. `false`: tidak.
`Table`	Informasi tabel sumber.
`filter`	Kondisi filter yang digunakan untuk memfilter data yang akan dimigrasikan, disinkronkan, atau dilanggan. Parameter ini hanya dapat diatur pada level tabel. Misalnya, Anda dapat mengatur parameter ini menjadi `id>10` untuk hanya memigrasikan atau menyinkronkan data yang nilainya pada kolom ID lebih besar dari 10. Untuk informasi lebih lanjut tentang format kondisi filter, lihat Atur kondisi filter. Catatan Tugas langganan tidak mendukung pengaturan kondisi filter.
`column`	Informasi kolom sumber.
`key`	Menentukan apakah kolom tersebut merupakan kunci primer. Nilai yang valid: `PRI`: ya. String kosong: tidak.
`sharedKey`	Menentukan apakah kolom tersebut merupakan kunci shard. Nilai yang valid: `true`: ya. `false`: tidak. Catatan Parameter ini wajib hanya ketika tipe database dari objek migrasi atau sinkronisasi adalah Kafka.
`type`	Tipe data bidang.
`state`	Jika nilainya `checked`, kolom tersebut dipilih.
`shard`	Jumlah shard untuk tabel yang akan dimigrasikan atau disinkronkan. Catatan Parameter ini wajib hanya ketika tipe database dari data migrasi atau sinkronisasi adalah Kafka.
`dml_op`	Operasi DML yang akan dimigrasikan atau disinkronkan secara inkremental. Nilai yang valid: `i`: INSERT. `u`: UPDATE. `d`: DELETE. Jika parameter ini kosong, semua operasi DML yang didukung oleh tugas akan dimigrasikan atau disinkronkan secara inkremental. `none`: Tidak ada operasi DML yang dimigrasikan atau disinkronkan secara inkremental. Catatan Untuk mengkueri operasi DML yang didukung oleh berbagai tugas migrasi atau sinkronisasi, lihat dokumen konfigurasi untuk tugas tertentu di Solusi migrasi atau Solusi sinkronisasi.
`ddl_op`	Operasi DDL yang akan dimigrasikan atau disinkronkan secara inkremental. Nilai yang valid: `ct`: CREATE TABLE. `at`: ALTER TABLE. `dt`: DROP TABLE. `rt`: RENAME TABLE. `tt`: TRUNCATE TABLE. Jika parameter ini kosong, semua operasi DDL yang didukung oleh tugas akan dimigrasikan atau disinkronkan secara inkremental. `none`: Tidak ada operasi DDL yang dimigrasikan atau disinkronkan secara inkremental. Catatan Untuk mengkueri operasi DDL yang didukung oleh berbagai tugas migrasi atau sinkronisasi, lihat dokumen konfigurasi untuk tugas tertentu di Solusi migrasi atau Solusi sinkronisasi.
`primary_key`	Menentukan kunci primer. Parameter ini tersedia dan wajib hanya ketika instans tujuan adalah AnalyticDB for MySQL atau AnalyticDB for PostgreSQL.
`part_key`	Menentukan kunci partisi. Parameter ini tersedia dan wajib ditentukan ketika instans tujuan adalah AnalyticDB for MySQL atau AnalyticDB for PostgreSQL.
`type` Penting Parameter `type` ini berbeda dari parameter `type` yang merepresentasikan tipe data suatu bidang.	Ketika instans tujuan adalah AnalyticDB for MySQL atau AnalyticDB for PostgreSQL, Anda perlu menentukan jenis tabel dari objek yang akan dimigrasikan atau disinkronkan: dimension: tabel dimensi. partition: tabel partisi.
`tagColumnValue`	Nilai kustom kolom tag __dts_data_source. Ketika instans tujuan adalah AnalyticDB for MySQL, parameter ini tersedia dan wajib diisi.
`conflict`	Kebijakan resolusi konflik global pada level tugas. Parameter ini harus disertakan dalam setiap database yang akan disinkronkan, dan nilainya harus sama. Nilai yang valid: `overwrite`: Saat terjadi konflik sinkronisasi data, catatan yang bertentangan di database tujuan langsung ditimpa. `interrupt`: Saat terjadi konflik sinkronisasi data, tugas sinkronisasi melaporkan kesalahan dan keluar. Tugas sinkronisasi masuk ke status gagal, dan Anda perlu melakukan intervensi untuk memperbaiki tugas. Catatan Konsol menampilkan TaskFailed. `ignore`: Saat terjadi konflik sinkronisasi data, pernyataan sinkronisasi saat ini dilewati, eksekusi dilanjutkan, dan catatan yang bertentangan di database tujuan digunakan.
`resolve_method`	Kebijakan resolusi konflik independen pada level tabel (hanya didukung untuk sinkronisasi inkremental). Nilai yang valid: `overwrite`: Saat terjadi konflik sinkronisasi data, catatan yang bertentangan di database tujuan langsung ditimpa. `interrupt`: Saat terjadi konflik sinkronisasi data, tugas sinkronisasi melaporkan kesalahan dan keluar. Tugas sinkronisasi masuk ke status gagal, dan Anda perlu melakukan intervensi untuk memperbaiki tugas. Catatan Konsol menampilkan TaskFailed. `ignore`: Saat terjadi konflik sinkronisasi data, pernyataan sinkronisasi saat ini dilewati, eksekusi dilanjutkan, dan catatan yang bertentangan di database tujuan digunakan. `use_max`: Saat terjadi konflik sinkronisasi data, kedua catatan yang bertentangan dibandingkan, dan catatan dengan nilai lebih besar ditulis ke database tujuan. Jika catatan tujuan tidak ada atau tipe bidang tidak memenuhi persyaratan, metode pemrosesan sistem setara dengan `overwrite`. `use_min`: Saat terjadi konflik sinkronisasi data, kedua catatan yang bertentangan dibandingkan, dan catatan dengan nilai lebih kecil ditulis ke database tujuan. Jika catatan tujuan tidak ada atau tipe bidang tidak memenuhi persyaratan, metode pemrosesan sistem setara dengan `ignore`.
`cdr_cmp_col`	Level tabel: Kolom deteksi konflik yang perlu diatur kebijakan resolusi konflik independennya, tidak termasuk kunci primer dan kunci unik. Nilainya harus sama. Penting Saat kebijakan resolusi konflik adalah `use_max` atau `use_min`, parameter `cdr_cmp_col` dan `cdr_rslv_col` harus ditentukan. Nilai sudah mencakup kunci primer dan kunci unik secara default. Anda tidak perlu menentukan kolom yang sesuai secara manual. Level database: Kolom deteksi konflik yang perlu diatur kebijakan resolusi konflik independennya. Nilainya harus sama. Penting Parameter `cdr_cmp_col` dan `cdr_rslv_col` harus ditentukan.
`cdr_rslv_col`

Contoh konfigurasi objek migrasi, sinkronisasi, atau langganan

Contoh 1: Migrasikan, sinkronkan, atau langgankan semua tabel di database dtstestdata.
```
{"dtstestdata": {   "name": "dtstestdata",   "all": true }}
```
Contoh 2: Migrasikan atau sinkronkan database dtstestdata dan ubah namanya menjadi dtstestdata_new.
```
{"dtstestdata": {   "name": "dtstestdata_new",   "all": true }}
```

Contoh 3: Migrasikan, sinkronkan, atau langgankan tabel tertentu (seperti customer) di database dtstestdata.

{"dtstestdata": {
   "name": "dtstestdata",
   "all": false,
   "Table": {
     "customer": {
       "name": "customer",
       "all": true, 
       "column": { 
         "id": {
           "key": "PRI",
           "name": "id",
           "type": "int(11)",
           "sharedKey": false,
           "state": "checked"  
         },
         "gmt_create": {
           "key": "",
           "name": "gmt_create",
           "type": "datetime",
           "sharedKey": false,
           "state": "checked"
         },
         "gmt_modify": {
           "key": "",
           "name": "gmt_modify",
           "type": "datetime",
           "sharedKey": false,
           "state": "checked"
         },
         "valid_time": {
           "key": "",
           "name": "valid_time",
           "type": "datetime",
           "sharedKey": false,
           "state": "checked"
         },
         "creator": {
           "key": "",
           "name": "creator",
           "type": "varchar(200)",
           "sharedKey": false,
           "state": "checked"
         }
       },
       "shard": 12
     }
   }
 }
}

Contoh 4: Migrasikan atau sinkronkan kolom tertentu dari tabel (seperti customer dan order) di database dtstestdata.

{"dtstestdata": {
   "name": "dtstestdata",
   "all": false,
   "Table": {
     "customer": {
       "name": "customer",
       "all": false, 
       "column": { 
         "id": {
           "key": "PRI",
           "name": "id",
           "type": "int(11)",
           "sharedKey": false,
           "state": "checked"  
         },
         "level": {
           "key": "",
           "name": "level",
           "type": "varchar(5000)",
           "sharedKey": false,
           "state": "checked"
         },
         "name": {
           "key": "",
           "name": "name",
           "type": "varchar(500)",
           "sharedKey": false,
           "state": "checked"
         },
       },
       "shard": 12
     },
     "order": {
       "name": "order",
       "all": false,
      "column": {
         "id": {
           "key": "PRI",
           "name": "id",
           "type": "int(11)",
           "sharedKey": false,
           "state": "checked"
         }
       },
       "shard": 12
     }
   }
 }
}

Contoh 5: Migrasikan atau sinkronkan tabel (seperti customer, order, dan commodity) dari database dtstestdata ke instans tujuan AnalyticDB for MySQL atau AnalyticDB for PostgreSQL.

{
    "dtstestdata": {
        "name": "dtstestdatanew",
        "all": false,
        "Table": {
            "order": {
                "name": "ordernew",
                "all": true,
                "part_key": "id",
                "primary_key": "id",
                "type": "partition"
            },
            "customer": {
                "name": "customernew",
                "all": true,
                "primary_key": "id",
                "type": "dimension"
            },
            "commodity": {
                "name": "commoditynew",
                "all": false,
                "filter": "id>10",
                "column": {
                    "id": {
                        "key": "PRI",
                        "name": "id",
                        "type": "int(11)"
                    }
                },
                "part_key": "id",
                "primary_key": "id",
                "type": "partition"
            }
        }
    }
}

Contoh 6: Tetapkan kebijakan resolusi konflik independen untuk objek sinkronisasi.

Pengaturan level tabel

Tetapkan kebijakan resolusi konflik global untuk objek tugas sinkronisasi menjadi interrupt. Tetapkan kebijakan resolusi konflik independen untuk kolom kunci primer, kolom kunci unik, dan kolom name dari tabel customer di database dtstestdata2 menjadi overwrite.

{
    "dtstestdata1": {
      "name": "dtstestdata1",
      "all": true,
      "conflict": "interrupt"
    },
    "dtstestdata2": {
      "name": "dtstestdata2",
      "all": false,
      "conflict": "interrupt",
      "Table": {
        "customer": {
          "name": "customer",
          "all": true,
          "cdr_cmp_col": "name",
          "cdr_rslv_col": "name",
          "resolve_method": "overwrite"
        }
      }
    }
  }

Pengaturan level database

Saat objek sinkronisasi adalah seluruh database: Tetapkan kebijakan resolusi konflik independen untuk kolom name dan addr dari semua tabel yang akan disinkronkan di database dtstestdata1 menjadi use_max.
```
"dtstestdata1": {
  "name": "dtstestdata1",
  "all": true,
  "conflict": "overwrite",
  "cdr_cmp_col": "name,addr",
  "cdr_rslv_col": "name,addr",
  "resolve_method": "use_max"
  }
}
```

Saat objek sinkronisasi bukan seluruh database: Tetapkan kebijakan resolusi konflik independen untuk kolom name dan addr dari semua tabel yang akan disinkronkan di database dtstestdata2 menjadi use_max.

"dtstestdata2": {
  "name": "dtstestdata2",
  "all": false,
  "conflict": "overwrite",
  "cdr_cmp_col": "name,addr",
  "cdr_rslv_col": "name,addr",
  "resolve_method": "use_max",
  "Table": {
    "person": {
      "name": "person",
      "all": true
    },
    "class": {
      "name": "class",
      "all": true
    }
  }
}

Contoh 7: Untuk instansi sinkronisasi dengan tipe database sumber Tair/Redis, sinkronkan hanya data dengan awalan kunci HProp dari DB bernama 0 dan 1 (yaitu, Prefixes of Keys to Be Synchronized adalah HProp). Untuk DB bernama 2, sinkronkan hanya data dengan awalan kunci dts tetapi tidak termasuk dtstest (yaitu, Prefixes of Keys to Be Synchronized adalah dts, dan Prefixes of Keys to Be Filtered Out adalah dtstest).

{
    "0": {
        "name": "0", 
        "all": true,
         "filter": "[{\"condition\":\"HProp\",\"filterType\":\"white\",\"filterPattern\":\"prefix\"}]"
    }, 
    "1": {
        "name": "1", 
        "all": true,
         "filter": "[{\"condition\":\"HProp\",\"filterType\":\"white\",\"filterPattern\":\"prefix\"}]"
    }, 
    "2": {
        "name": "2", 
        "all": true,
         "filter": "[{\"condition\":\"dts\",\"filterType\":\"white\",\"filterPattern\":\"prefix\"},{\"condition\":\"dtstest\",\"filterType\":\"black\",\"filterPattern\":\"prefix\"}]"
    }
}

Contoh 8: Integrasikan tabel commodity di database dtstestdata ke OSS tujuan dalam format Delta.

{
    "dtstestdata(": {
        "all": false, 
        "Table": {
            "commodity": {
                "all": false(), 
                "filter": "", 
                "write_operation": "overwrite", 
                "name": "commodity", 
                "column": {
                    "IS_VALID": {
                        "name": "is_valid", 
                        "targetType": "String"
                    }, 
                    "BuiltinArchiveDate": {
                        "name": "builtinarchivedate", 
                        "targetType": "String"
                    }, 
                    "PRODUCT_NAME": {
                        "name": "product_name", 
                        "targetType": "String"
                    }, 
                    "PRODUCT_CODE": {
                        "name": "product_code", 
                        "targetType": "String"
                    }, 
                    "etl_date": {
                        "syntacticType": "ADD", 
                        "name": "etl_date", 
                        "type": "String", 
                        "value": "'2025-07-08 03:30:00'"
                    }
                }, 
                "part_key": "dt=2025-07-07"
            }
        }, 
        "name": "dtstestdata"
    }
}