MaxCompute：Pembaruan Kolom Parsial Hampir Real-Time ke Tabel Delta dengan Flink

Proses operasi

1. Seorang pengguna memperbarui nama dan nomor telepon mereka di platform berbeda. StreamA menerima pembaruan nama, dan StreamB menerima pembaruan nomor telepon.

2. Kedua StreamA dan StreamB mengirimkan pembaruan ke sistem manajemen informasi pengguna.

Hasil akhir

• Tanpa Pembaruan Kolom Parsial: Jika pembaruan dari StreamB tiba dan diproses setelah StreamA, itu akan menimpa informasi nama yang baru saja diperbarui oleh StreamA (jika StreamB melakukan pembaruan baris penuh), menyebabkan nama kembali ke nilai lamanya.

• Dengan Pembaruan Kolom Parsial:

  • Ketika StreamA melakukan pembaruan, itu hanya beroperasi pada kolom nama tanpa memengaruhi kolom informasi kontak.

  • Ketika StreamB melakukan pembaruan, itu hanya beroperasi pada kolom nomor telepon tanpa memengaruhi kolom informasi pribadi.

  Hasil akhirnya adalah nama pengguna diperbarui ke informasi terbaru, dan nomor telepon juga diperbarui ke informasi terbaru. Pembaruan ini dilakukan secara independen tanpa gangguan, memastikan integritas dan akurasi informasi pengguna.

Proses operasi

1. StreamA hanya ingin memperbarui usia pengguna, dengan perintah seperti: INSERT INTO table (pk, age) VALUES (1, 3);.

2. Pada saat yang sama, StreamB hanya ingin memperbarui jenis kelamin pengguna, dengan perintah seperti: INSERT INTO table (pk, sex) VALUES (1, 'male');.

Hasil akhir

• Tanpa Pembaruan Kolom Parsial: Jika rekaman dengan kunci utama 1 menerima perintah pembaruan di atas, semua bidang lain selain yang sedang diperbarui akan disetel ke NULL. Ini akan mengakibatkan hilangnya data valid asli.

• Dengan Pembaruan Dinamis:

  • Ketika operasi insert dipicu, sistem mengidentifikasi bahwa hanya beberapa bidang yang berisi data.

  • Mekanisme pembaruan kolom parsial memastikan bahwa hanya bidang-bidang dengan data yang diperbarui.

  • Pada saat yang sama, bidang yang tidak memiliki data yang diberikan dalam operasi insert akan mempertahankan nilai aslinya.

Parameter

Untuk mengonfigurasi mode pembaruan kolom parsial, MaxCompute telah memperkenalkan dua parameter konfigurasi berikut:

Contoh konfigurasi pembaruan kolom parsial dinamis

Buat tabel dengan pembaruan kolom parsial dinamis diaktifkan. Contoh:

CREATE TABLE partialtable (
  pk INT,
  c1 STRING, 
  c2 STRING, 
  c3 STRING,
  PRIMARY KEY(pk) NOT ENFORCED
) WITH (
  'connector' = 'maxcompute',
  'odps.end.point' = 'https://service.cn-hangzhou-vpc.maxcompute.aliyun-inc.com/api', //Konektivitas jaringan VPC
  'odps.project.name' = 'project_name',
  'odps.namespace.schema' = 'true', //Dukung model tiga lapis.
  'table.name' = 'project.schema.tablename',
  'sink.operation' = 'upsert',
  'upsert.write.bucket.num' = '1',
  'upsert.partial-column.enable' = 'true', 
  'odps.access.id' = 'yourAccessId',
  'odps.access.key' = 'yourAccessKey'
);

Contoh operasi berikutnya pada tabel dan hasilnya:

1. Masukkan data [1,a, b, c] ke dalam tabel, di mana kolom pertama adalah kunci utama. Data awal adalah [1, a, b, c].

   INSERT INTO partialtable VALUES (1, 'a', 'b', 'c'); 

2. Perbarui hanya kolom kedua c2 menjadi d untuk rekaman dengan kunci utama 1. Data setelah pembaruan adalah [1, a, d, c].

   INSERT INTO partialtable(pk, c2) VALUES (1, 'd'); 

3. Perbarui hanya kolom ketiga c3 menjadi e untuk rekaman dengan kunci utama 1. Data setelah pembaruan adalah [1, a, d, e].

   INSERT INTO partialtable(pk, c3) VALUES (1, 'e'); 

Contoh konfigurasi pembaruan kolom parsial statis

Buat tabel yang hanya memperbarui kolom c2. Operasi berikutnya pada tabel ini hanya akan memengaruhi kolom c2, sementara kolom lainnya akan tetap tidak berubah. Contoh:

CREATE TABLE PartialTable2 (
  pk INT,
  c1 STRING, 
  c2 STRING, 
  c3 STRING,
  PRIMARY KEY(pk) NOT ENFORCED
) WITH (
  'connector' = 'maxcompute',
  'odps.end.point' = 'https://service.cn-hangzhou-vpc.maxcompute.aliyun-inc.com/api', //Konektivitas jaringan VPC
  'odps.project.name' = 'project_name',
  'odps.namespace.schema' = 'true', //Dukung model tiga lapis.
  'table.name' = 'project.schema.tablename',
  'sink.operation' = 'upsert',
  'upsert.write.bucket.num' = '1',
  'upsert.partial-column.enable' = 'true', 
  'upsert.partial-column.name' = 'c2', // Tentukan untuk hanya memperbarui kolom c2.
  'odps.access.id' = 'yourAccessId',
  'odps.access.key' = 'yourAccessKey'
);