Tablestore：Sinkronisasi data inkremental ke OSS

Langkah 1: Tambahkan sumber data Tablestore

Konfigurasikan sumber data Tablestore di DataWorks untuk menghubungkan ke data sumber.

1. Login ke Konsol DataWorks. Alihkan ke wilayah tujuan. Di panel navigasi kiri, pilih Data Integration > Data Integration. Dari daftar drop-down, pilih ruang kerja dan klik Go To Data Integration.

2. Di panel navigasi kiri, klik Data source.

3. Di halaman Data Sources, klik Add Data Source.

4. Di kotak dialog Add Data Source, cari dan pilih Tablestore sebagai tipe sumber data.

5. Di kotak dialog Add OTS Data Source, konfigurasikan parameter sumber data seperti dijelaskan dalam tabel berikut.

   Parameter	Deskripsi
   Data Source Name	Nama sumber data harus merupakan kombinasi huruf, angka, dan garis bawah (_). Nama tidak boleh diawali dengan angka atau garis bawah (_).
   Data Source Description	Deskripsi singkat mengenai sumber data. Panjang deskripsi tidak boleh melebihi 80 karakter.
   Region	Pilih wilayah tempat instans Tablestore berada.
   Tablestore Instance Name	Nama instans Tablestore.
   Endpoint	Titik akhir (endpoint) instans Tablestore. Gunakan alamat VPC.
   AccessKey ID	ID AccessKey dan Rahasia AccessKey dari Akun Alibaba Cloud atau pengguna RAM.
   AccessKey Secret

6. Uji konektivitas kelompok sumber daya.

   Saat membuat sumber data, Anda harus menguji konektivitas kelompok sumber daya untuk memastikan bahwa kelompok sumber daya untuk tugas sinkronisasi dapat terhubung ke sumber data. Jika tidak, tugas sinkronisasi data tidak dapat dijalankan.

   1. Di bagian Connection Configuration, klik Test Network Connectivity di kolom Connection Status untuk kelompok sumber daya.

   2. Setelah uji konektivitas berhasil, klik Complete. Sumber data baru akan muncul dalam daftar sumber data.

      Jika uji konektivitas gagal, gunakan Network Connectivity Diagnostic Tool untuk memecahkan masalah.

Langkah 2: Tambahkan sumber data OSS

Konfigurasikan sumber data OSS sebagai tujuan ekspor data.

1. Klik Add Data Source lagi. Di kotak dialog, cari dan pilih OSS sebagai tipe sumber data, lalu konfigurasikan parameter sumber data.

   Parameter	Deskripsi
   Data Source Name	Nama sumber data harus terdiri dari huruf, angka, dan garis bawah (_). Nama tidak boleh diawali dengan angka atau garis bawah (_).
   Data Source Description	Deskripsi singkat mengenai sumber data. Panjang deskripsi tidak boleh melebihi 80 karakter.
   Access Mode	RAM Role Authorization Mode: Akun layanan DataWorks mengakses sumber data dengan mengasumsikan peran RAM. Jika ini pertama kalinya Anda memilih mode ini, ikuti petunjuk di layar untuk memberikan izin yang diperlukan. AccessKey Mode: Akses sumber data menggunakan ID AccessKey dan Rahasia AccessKey dari Akun Alibaba Cloud atau pengguna RAM.
   Role	Parameter ini hanya diperlukan jika Anda mengatur Access Mode ke RAM Role Authorization Mode.
   AccessKey ID	Parameter ini hanya diperlukan jika Anda mengatur Access Mode ke AccessKey Mode. ID AccessKey dan Rahasia AccessKey dari Akun Alibaba Cloud atau pengguna RAM.
   AccessKey Secret
   Region	Wilayah tempat bucket berada.
   Endpoint	Nama domain OSS. Untuk informasi lebih lanjut, lihat Regions and endpoints.
   Bucket	Nama bucket.

2. Setelah Anda mengonfigurasi parameter dan uji konektivitas berhasil, klik Complete untuk menambahkan sumber data.

Langkah 3: Konfigurasikan tugas sinkronisasi batch

Buat dan konfigurasikan tugas sinkronisasi data untuk menentukan aturan transfer data dari Tablestore ke OSS.

{
    "type": "job",
    "version": "2.0",
    "steps": [
        {
            "stepType": "otsstream",
            "parameter": {
                "statusTable": "TableStoreStreamReaderStatusTable",
                "maxRetries": 31,
                "isExportSequenceInfo": false,
                "datasource": "source_data",
                "column": [
                    "id",
                    "name",
                    "colName",
                    "version",
                    "colValue",
                    "opType",
                    "sequenceInfo"
                ],
                "startTimeString": "${startTime}",
                "table": "source_table",
                "endTimeString": "${endTime}"
            },
            "name": "Reader",
            "category": "reader"
        },
        {
            "stepType": "oss",
            "parameter": {
                "dateFormat": "yyyy-MM-dd HH:mm:ss",
                "datasource": "target_data",
                "writeSingleObject": false,
                "column": [
                    "0",
                    "1",
                    "2",
                    "3",
                    "4",
                    "5",
                    "6"
                ],
                "writeMode": "truncate",
                "encoding": "UTF-8",
                "fieldDelimiter": ",",
                "fileFormat": "csv",
                "object": "tablestore/source_table.csv"
            },
            "name": "Writer",
            "category": "writer"
        }
    ],
    "setting": {
        "errorLimit": {
            "record": "0"
        },
        "speed": {
            "concurrent": 2,
            "throttle": false
        }
    },
    "order": {
        "hops": [
            {
                "from": "Reader",
                "to": "Writer"
            }
        ]
    }
}

{
    "type": "job",
    "version": "2.0",
    "steps": [
        {
            "stepType": "otsstream",
            "parameter": {
                "statusTable": "TableStoreStreamReaderStatusTable",
                "maxRetries": 31,
                "isExportSequenceInfo": false,
                "datasource": "source_data",
                "column": [
                  {
                    "name": "_m_name"
                  },
                  {
                    "name": "_data_source"
                  },
                  {
                    "name": "_tags"
                  },
                  {
                    "name": "_time"
                  },
                  {
                    "name": "value",
                    "type": "int"
                  }
                ],
                "startTimeString": "${startTime}",
                "table": "source_series",
                "isTimeseriesTable":"true",
                "mode": "single_version_and_update_only",
                "endTimeString": "${endTime}"
            },
            "name": "Reader",
            "category": "reader"
        },
        {
            "stepType": "oss",
            "parameter": {
                "dateFormat": "yyyy-MM-dd HH:mm:ss",
                "datasource": "target_data",
                "writeSingleObject": false,
                "writeMode": "truncate",
                "encoding": "UTF-8",
                "fieldDelimiter": ",",
                "fileFormat": "csv",
                "object": "tablestore/source_series.csv"
            },
            "name": "Writer",
            "category": "writer"
        }
    ],
    "setting": {
        "errorLimit": {
            "record": "0"
        },
        "speed": {
            "concurrent": 2,
            "throttle": false
        }
    },
    "order": {
        "hops": [
            {
                "from": "Reader",
                "to": "Writer"
            }
        ]
    }
}

Langkah 4: Lihat hasil sinkronisasi

Setelah tugas sinkronisasi selesai, Anda dapat melihat status eksekusi di log dan memeriksa file hasil di bucket OSS.

1. Lihat status dan hasil eksekusi tugas di bagian bawah halaman. Informasi log berikut menunjukkan bahwa tugas sinkronisasi berhasil dijalankan.

   2025-11-18 11:16:23 INFO Shell run successfully!
   2025-11-18 11:16:23 INFO Current task status: FINISH
   2025-11-18 11:16:23 INFO Cost time is: 77.208s

2. Lihat file di bucket tujuan.

   Buka Daftar Bucket. Klik bucket tujuan untuk melihat atau mengunduh file hasil.