DataWorks：Sumber Data Stream Tablestore

Siapkan lingkungan Tablestore Stream sebelum sinkronisasi data

SyncClient client = new SyncClient("", "", "", "");
#Metode 1: Aktifkan fitur Stream saat Anda membuat tabel.
CreateTableRequest createTableRequest = new CreateTableRequest(tableMeta);
createTableRequest.setStreamSpecification(new StreamSpecification(true, 24)); // Nilai 24 menunjukkan bahwa Tablestore menyimpan data inkremental selama 24 jam.
client.createTable(createTableRequest);
#Metode 2: Jika Anda tidak mengaktifkan fitur Stream saat membuat tabel, Anda dapat memanggil operasi UpdateTable untuk mengaktifkan fitur ini setelah tabel dibuat.
UpdateTableRequest updateTableRequest = new UpdateTableRequest("tableName");
updateTableRequest.setStreamSpecification(new StreamSpecification(true, 24));
client.updateTable(updateTableRequest);

Mode baca dan tipe data yang didukung

Tablestore Stream Reader memungkinkan Anda membaca data inkremental dari Tablestore dalam mode kolom atau baris. Bagian ini menjelaskan proses pembacaan data dan tipe data yang didukung oleh mode baca tersebut.

Membaca data dalam mode kolom

Dalam mode multi-versi Tablestore, data tabel diorganisasikan dalam struktur tiga tingkat: baris, kolom, dan versi. Satu baris dapat memiliki beberapa kolom dengan nama kolom yang tidak tetap. Setiap kolom dapat memiliki beberapa versi, dan setiap versi memiliki timestamp tertentu sebagai nomor versi.

Anda dapat memanggil operasi API Tablestore untuk membaca dan menulis data. Tablestore mencatat operasi tulis dan modifikasi terbaru pada data tabel untuk mencatat data inkremental. Oleh karena itu, data inkremental dapat dianggap sebagai kumpulan catatan operasi.

Tablestore mendukung kolom dinamis dan mode multi-versi. Oleh karena itu, sebuah baris yang dihasilkan oleh Tablestore Stream Reader adalah versi dari sebuah kolom, bukan baris di Tablestore. Setelah Tablestore Stream Reader membaca data dari sebuah baris di Tablestore, data tersebut dikonversi menjadi beberapa baris. Setiap baris mencakup nilai kunci utama, nama kolom, timestamp versi untuk kolom (nomor versi), nilai versi, dan tipe operasi. Jika parameter isExportSequenceInfo disetel ke true, informasi deret waktu juga disertakan.

Membaca data dalam mode baris

• Tabel Lebar

  Anda juga dapat menggunakan Tablestore Stream Reader untuk membaca data dalam mode baris. Dalam mode ini, Tablestore Stream Reader membaca catatan operasi sebagai baris. Anda harus mengonfigurasi parameter mode dan menentukan kolom-kolom yang ingin dibaca.

  "parameter": {
    # Setel mode ke single_version_and_update_only, setel isExportSequenceInfo ke false, dan konfigurasikan parameter lainnya, seperti datasource dan table, berdasarkan kebutuhan bisnis Anda.
    "mode": "single_version_and_update_only", # Mode baca.
    "column":[  # Kolom-kolom yang ingin Anda baca datanya dari Tablestore. Anda dapat menentukan kolom berdasarkan kebutuhan bisnis Anda.
            {
               "name": "uid"  # Nama kolom, yang bisa berupa kolom kunci utama atau kolom properti.
            },
            {
               "name": "name"  # Nama kolom, yang bisa berupa kolom kunci utama atau kolom properti.
            },
    ],
    "isExportSequenceInfo": false, # Menentukan apakah membaca informasi deret waktu. Jika Anda menyetel parameter mode ke single_version_and_update_only, parameter ini hanya dapat disetel ke false.
  }

• Tabel Seri Waktu

  Fitur Stream secara otomatis diaktifkan saat Anda membuat tabel seri waktu.

  Tablestore Stream Reader memungkinkan Anda membaca data inkremental dari tabel seri waktu. Jika tabel tersebut merupakan tabel seri waktu, Anda harus mengonfigurasi parameter berikut:

  "parameter": {
    # Konfigurasikan parameter berikut dan parameter lainnya, seperti datasource dan table, berdasarkan kebutuhan bisnis Anda.
    "mode": "single_version_and_update_only", # Mode baca.
    "isTimeseriesTable":"true", # Menentukan apakah tabel tersebut merupakan tabel seri waktu.
    "column":[  # Kolom-kolom yang ingin Anda baca datanya dari Tablestore. Anda dapat menentukan kolom berdasarkan kebutuhan bisnis Anda.
            {
              "name": "_m_name" # Nama kolom nama metrik.
            },
            {
              "name": "_data_source" # Nama kolom sumber data.
            },
            {
              "name": "_tags" # Nama kolom tag. Tag diubah menjadi data tipe string.
            },
            {
              "name": "tag1", # Nama kunci tag.
              "is_timeseries_tag":"true"  # Menentukan apakah bidang tersebut merupakan bidang internal dari tag.
            },
            {
              "name": "time" # Nama kolom timestamp.
            },
            {
               "name": "name" # Nama kolom properti.
            },
    ],
    "isExportSequenceInfo": false, # Menentukan apakah membaca informasi deret waktu. Jika Anda menyetel parameter mode ke single_version_and_update_only, parameter ini hanya dapat disetel ke false.
  }

  Data yang dibaca per baris lebih dekat dengan data dalam baris aslinya, memudahkan pemrosesan data lebih lanjut. Jika Anda membaca data per baris, perhatikan poin-poin berikut:

  • Baris-baris yang dibaca diekstraksi dari catatan operasi. Setiap baris sesuai dengan operasi tulis atau pembaruan. Jika Anda hanya memperbarui beberapa kolom untuk sebuah baris, catatan operasi hanya berisi kolom-kolom yang diperbarui.

  • Nomor versi setiap kolom, yaitu timestamp setiap kolom, tidak dapat dibaca atau dihapus.

Pemetaan tipe data

Mengembangkan tugas sinkronisasi data: Panduan prosedur untuk sinkronisasi inkremental dari Tablestore

• Untuk informasi tentang prosedur konfigurasi, lihat Konfigurasikan Tugas Sinkronisasi Batch Menggunakan UI Tanpa Kode dan Konfigurasikan Tugas Sinkronisasi Batch Menggunakan Editor Kode.

• Untuk informasi tentang semua parameter yang dikonfigurasi dan kode yang dijalankan saat menggunakan editor kode untuk mengonfigurasi tugas sinkronisasi batch, lihat Lampiran: Kode dan Parameter.

Lampiran: Kode dan parameter

Lampiran: Konfigurasikan tugas sinkronisasi batch menggunakan editor kode

Kode untuk Tablestore Stream Reader

• Membaca Data dalam Mode Kolom

  {
      "type":"job",
      "version":"2.0",// Nomor versi.
      "steps":[
          {
              "stepType":"otsstream",// Nama plugin.
              "parameter":{
                  "datasource":"$srcDatasource",// Nama sumber data.
                  "dataTable":"",// Nama tabel.
                  "statusTable":"TableStoreStreamReaderStatusTable",// Nama tabel yang digunakan oleh Tablestore Stream Reader untuk menyimpan catatan status.
                  "maxRetries":30,// Jumlah maksimum percobaan ulang untuk setiap permintaan membaca data inkremental dari Tablestore. Nilai default: 30.
                  "isExportSequenceInfo":false,// Menentukan apakah membaca informasi deret waktu.
                  "startTimeString":"${startTime}${hh}",// Waktu mulai data inkremental, yang sama dengan waktu ketika tugas sinkronisasi mulai dijalankan. Konfigurasikan parameter ini dalam format yyyymmddhh24miss.
                  "endTimeString":"${endTime}${hh}"// Waktu akhir data inkremental, yang sama dengan waktu ketika tugas sinkronisasi selesai dijalankan. Konfigurasikan parameter ini dalam format yyyymmddhh24miss.
              },
              "name":"Reader",
              "category":"reader"
          },
          {
              "stepType":"stream",
              "parameter":{},
              "name":"Writer",
              "category":"writer"
          }
      ],
      "setting":{
          "errorLimit":{
              "record":"0"// Jumlah maksimum catatan data kotor yang diizinkan.
          },
          "speed":{
              "throttle":true,// Menentukan apakah throttling diaktifkan. Nilai false menunjukkan bahwa throttling dinonaktifkan, dan nilai true menunjukkan bahwa throttling diaktifkan. Parameter mbps hanya berlaku saat parameter throttle disetel ke true.
              "concurrent":1,// Jumlah maksimum thread paralel.
              "mbps":"12"// Laju transmisi maksimum. Unit: MB/s.
  
          }
      },
      "order":{
          "hops":[
              {
                  "from":"Reader",
                  "to":"Writer"
              }
          ]
      }
  }

• Membaca Data dari Tabel Lebar dalam Mode Baris

  {
      "type":"job",
      "version":"2.0",// Nomor versi.
      "steps":[
          {
              "stepType":"otsstream",// Nama plugin.
              "parameter":{
                  "datasource":"$srcDatasource",// Nama sumber data.
                  "dataTable":"",// Nama tabel.
                  "statusTable":"TableStoreStreamReaderStatusTable",// Nama tabel yang digunakan oleh Tablestore Stream Reader untuk menyimpan catatan status.
                  "maxRetries":30,// Jumlah maksimum percobaan ulang untuk setiap permintaan membaca data inkremental dari Tablestore. Nilai default: 30.
                  "isExportSequenceInfo":false,// Menentukan apakah membaca informasi deret waktu.
                  "startTimeString":"${startTime}${hh}",// Waktu mulai data inkremental, yang sama dengan waktu ketika tugas sinkronisasi mulai dijalankan. Konfigurasikan parameter ini dalam format yyyymmddhh24miss.
                  "endTimeString":"${endTime}${hh}"// Waktu akhir data inkremental, yang sama dengan waktu ketika tugas sinkronisasi selesai dijalankan. Konfigurasikan parameter ini dalam format yyyymmddhh24miss.
                  "mode": "single_version_and_update_only",
                  "column":[
                          {
                              "name":"pId"
                          },
                          {
                              "name": "uId"
                          },
                          {
                              "name":"col0"
                          },
                          {
                              "name": "col1"
                          }
                      ],
                      },
              "name":"Reader",
              "category":"reader"
          },
          {
              "stepType":"stream",
              "parameter":{},
              "name":"Writer",
              "category":"writer"
          }
      ],
      "setting":{
          "errorLimit":{
              "record":"0"// Jumlah maksimum catatan data kotor yang diizinkan.
          },
          "speed":{
              "throttle":true,// Menentukan apakah throttling diaktifkan. Nilai false menunjukkan bahwa throttling dinonaktifkan, dan nilai true menunjukkan bahwa throttling diaktifkan. Parameter mbps hanya berlaku saat parameter throttle disetel ke true.
              "concurrent":1,// Jumlah maksimum thread paralel.
              "mbps":"12" // Laju transmisi maksimum.
  
          }
      },
      "order":{
          "hops":[
              {
                  "from":"Reader",
                  "to":"Writer"
              }
          ]
      }
  }

• Membaca Data dari Tabel Seri Waktu dalam Mode Baris

  {
      "type":"job",
      "version":"2.0",// Nomor versi.
      "steps":[
          {
              "stepType":"otsstream",// Nama plugin.
              "parameter":{
                  "datasource":"$srcDatasource",// Nama sumber data.
                  "dataTable":"",// Nama tabel.
                  "statusTable":"TableStoreStreamReaderStatusTable",// Nama tabel yang digunakan oleh Tablestore Stream Reader untuk menyimpan catatan status.
                  "maxRetries":30,// Jumlah maksimum percobaan ulang untuk setiap permintaan membaca data inkremental dari Tablestore. Nilai default: 30.
                  "isExportSequenceInfo":false,// Menentukan apakah membaca informasi deret waktu.
                  "startTimeString":"${startTime}${hh}",// Waktu mulai data inkremental, yang sama dengan waktu ketika tugas sinkronisasi mulai dijalankan. Konfigurasikan parameter ini dalam format yyyymmddhh24miss.
                  "endTimeString":"${endTime}${hh}"// Waktu akhir data inkremental, yang sama dengan waktu ketika tugas sinkronisasi selesai dijalankan. Konfigurasikan parameter ini dalam format yyyymmddhh24miss.
                  "mode": "single_version_and_update_only",
                  "isTimeseriesTable":"true",
                  "column": [
                            {
                              "name": "_m_name"
                            },
                            {
                              "name": "_data_source",
                            },
                            {
                              "name": "_tags",
                            },
                            {
                              "name": "string_column",
                            }
                      ]
                      },
              "name":"Reader",
              "category":"reader"
          },
          {
              "stepType":"stream",
              "parameter":{},
              "name":"Writer",
              "category":"writer"
          }
      ],
      "setting":{
          "errorLimit":{
              "record":"0"// Jumlah maksimum catatan data kotor yang diizinkan.
          },
          "speed":{
              "throttle":true,// Menentukan apakah throttling diaktifkan. Nilai false menunjukkan bahwa throttling dinonaktifkan, dan nilai true menunjukkan bahwa throttling diaktifkan. Parameter mbps hanya berlaku saat parameter throttle disetel ke true.
              "concurrent":1,// Jumlah maksimum thread paralel.
              "mbps":"12"// Laju transmisi maksimum. Unit: MB/s.
  
          }
      },
      "order":{
          "hops":[
              {
                  "from":"Reader",
                  "to":"Writer"
              }
          ]
      }
  }

Parameter dalam kode untuk Tablestore Stream Reader

"column":[
    {"name":"pk1"},
	{"name":"col1"},
	{"name":"col2","dataType":"new"},
	{"name":"col2","dataType":"old"},
	{"name":"col2","dataType":"latest"}
],