Data Transmission Service：Konfigurasikan Tugas ETL dalam Mode Flink SQL

Informasi Latar Belakang

• Sebelum mengonfigurasi tugas ETL, perhatikan informasi berikut:

  • Input/Dimension Table menunjukkan database sumber dari tugas ETL.

  • Output menunjukkan database tujuan dari tugas ETL.

• DTS menyediakan fitur ETL streaming untuk proses sinkronisasi data. Anda dapat menambahkan berbagai komponen transformasi antara database sumber dan tujuan untuk mentransformasi data dan menulis data yang diproses ke database tujuan secara real-time. Sebagai contoh, Anda dapat menggabungkan dua tabel aliran menjadi satu tabel besar dan menulis data dari tabel besar tersebut ke database tujuan. Anda juga dapat menambahkan bidang ke tabel sumber dan mengonfigurasi fungsi untuk memberi nilai pada bidang tersebut. Kemudian, Anda dapat menulis bidang tersebut ke database tujuan.

Prasyarat

• Tugas ETL dibuat di salah satu wilayah berikut: China (Hangzhou), China (Shanghai), China (Qingdao), China (Beijing), China (Zhangjiakou), China (Shenzhen), China (Guangzhou), dan China (Hong Kong).

• Database sumber mencakup salah satu jenis berikut: MySQL, PolarDB for MySQL, Oracle, PostgreSQL, iSeries DB2 (AS/400), Db2 for LUW, PolarDB-X 1.0, PolarDB for PostgreSQL, MariaDB, PolarDB for Oracle, SQL Server, dan PolarDB-X 2.0.

• Database tujuan mencakup salah satu jenis berikut: MySQL, PolarDB for MySQL, Oracle, AnalyticDB for MySQL V3.0, PolarDB for PostgreSQL, PostgreSQL, Db2 for LUW, iSeries DB2 (AS/400), AnalyticDB for PostgreSQL, SQL Server, MariaDB, PolarDB-X 1.0, PolarDB for Oracle, dan Tablestore.

• Skema tabel di database tujuan harus sudah dibuat karena fitur ETL tidak mendukung migrasi skema. Sebagai contoh, Tabel A berisi Bidang 1, Bidang 2, dan Bidang 3, sedangkan Tabel B berisi Bidang 2, Bidang 3, dan Bidang 4. Jika Anda ingin menggabungkan Tabel A dan Tabel B menjadi tabel yang berisi Bidang 2 dan Bidang 3, Anda harus membuat Tabel C yang berisi Bidang 2 dan Bidang 3 di database tujuan.

• Fitur ETL tidak mendukung sinkronisasi data penuh. Oleh karena itu, Anda hanya dapat mentransformasi data tambahan secara real-time.

Catatan Penggunaan

• Database sumber dan tujuan harus berada di wilayah yang sama.

• Semua stream tables harus milik instance yang sama.

• Semua nama database dan nama tabel harus unik.

• Database sumber dan tujuan harus berada dalam akun Alibaba Cloud yang sama.

Prosedur

1. Pergi ke halaman Streaming ETL.

   1. Masuk ke Konsol DTS.

   2. Di panel navigasi di sebelah kiri, klik ETL.

2. Klik . Di kotak dialog Create Data Flow, masukkan nama untuk tugas ETL di bidang Data Flow Name dan pilih FlinkSQL sebagai Development Method.

3. Klik OK.

4. Di bagian Data Flow Information dari halaman Streaming ETL, tambahkan database sumber dan tujuan.

   Parameter	Deskripsi
   Wilayah	Wilayah tempat database sumber atau tujuan berada.
   Tipe	Tipe tabel. Jika tabel sumber adalah tabel aliran (Tabel yang diperbarui secara real-time dan dapat dikaitkan dengan tabel dimensi untuk kueri asosiasi data), pilih Stream Table. Jika tabel sumber adalah tabel dimensi (Tabel yang tidak diperbarui secara real-time dan umumnya digunakan untuk merakit data real-time ke dalam tabel lebar untuk analisis data), pilih Dimension Table. Jika Anda mengonfigurasi tabel tujuan, pilih Output.
   Tipe Database	Tipe database sumber atau tujuan.
   Instance	Nama atau ID instance sumber atau tujuan. Penting Sebelum Anda menyetel parameter ini, Anda harus mendaftarkan instance sumber dan tujuan dengan Data Management (DMS). Untuk informasi lebih lanjut, lihat Manajemen Instance.
   Database	Database sumber atau tujuan tempat objek transformasi data berada.
   Tabel Fisik	Tabel sumber atau tujuan tempat objek transformasi data berada.
   Alias Tabel Fisik	Nama yang dapat dibaca dari tabel sumber atau tujuan. Nama ini membantu Anda mengidentifikasi tabel saat Anda mengeksekusi pernyataan SQL dalam ETL.

5. Di halaman Streaming ETL, masukkan pernyataan SQL di editor skrip guna mengonfigurasi tugas ETL.

   Pernyataan SQL berikut menunjukkan cara mengonfigurasi tugas ETL untuk menggabungkan tabel aliran bernama test_orders dengan tabel dimensi bernama product dan menyisipkan hasilnya ke tabel tujuan test_orders_new.

   Penting

   Setiap pernyataan SQL harus diakhiri dengan titik koma (;).

   CREATE TABLE `etltest_test_orders` (
     `order_id` BIGINT,
     `user_id` BIGINT,
     `product_id` BIGINT,
     `total_price` DECIMAL(15,2),
     `order_date` TIMESTAMP(6),
     `dts_etl_schema_db_table` STRING,
     `dts_etl_db_log_time` BIGINT,
     `pt` AS PROCTIME(),
     WATERMARK FOR `order_date` AS `order_date` - INTERVAL '5' SECOND
   ) WITH (
     'streamType'= 'append',
     'alias'= 'test_orders',
     'vertexType'= 'stream'
   );
   CREATE TABLE `etltest_product` (
     `product_id` BIGINT,
     `product_name` STRING,
     `product_price` DECIMAL(15,2)
   ) WITH (
     'alias'= 'product',
     'vertexType'= 'lookup'
   );
   CREATE VIEW `etltest_test_orders_JOIN_etltest_product` AS
   SELECT
     `etltest_test_orders`.`order_id` AS `order_id`,
     `etltest_test_orders`.`user_id` AS `user_id`,
     `etltest_test_orders`.`product_id` AS `product_id`,
     `etltest_test_orders`.`total_price` AS `total_price`,
     `etltest_test_orders`.`order_date` AS `order_date`,
     `etltest_test_orders`.`dts_etl_schema_db_table` AS `dts_etl_schema_db_table`,
     `etltest_test_orders`.`dts_etl_db_log_time` AS `dts_etl_db_log_time`,
     `etltest_product`.`product_id` AS `product_id_0001011101`,
     `etltest_product`.`product_name` AS `product_name`,
     `etltest_product`.`product_price` AS `product_price`
   FROM `etltest_test_orders` LEFT JOIN `etltest_product` FOR SYSTEM_TIME AS OF `etltest_test_orders`.`pt` ON etltest_test_orders.product_id = etltest_product.product_id
   ;
   CREATE TABLE `test_orders_new` (
     `order_id` BIGINT,
     `user_id` BIGINT,
     `product_id` BIGINT,
     `total_price` DECIMAL(15,2),
     `order_date` TIMESTAMP(6),
     `product_name` STRING,
     `product_price` DECIMAL(15,2)
   ) WITH (
     'alias'= 'test_orders_new',
     'vertexType'= 'sink'
   );
   INSERT INTO `test_orders_new` (
     `order_id`,
     `user_id`,
     `product_id`,
     `total_price`,
     `order_date`,
     `product_name`,
     `product_price`
   )
   SELECT
     `etltest_test_orders_JOIN_etltest_product`.`order_id`,
     `etltest_test_orders_JOIN_etltest_product`.`user_id`,
     `etltest_test_orders_JOIN_etltest_product`.`product_id`,
     `etltest_test_orders_JOIN_etltest_product`.`total_price`,
     `etltest_test_orders_JOIN_etltest_product`.`order_date`,
     `etltest_test_orders_JOIN_etltest_product`.`product_name`,
     `etltest_test_orders_JOIN_etltest_product`.`product_price`
   FROM `etltest_test_orders_JOIN_etltest_product`;

   Tipe	Deskripsi
   Tabel Sumber dan Tujuan	Anda harus mengeksekusi pernyataan CREATE TABLE untuk mendefinisikan tabel sumber dan tabel tujuan. Anda dapat menyetel tiga parameter di klausa WITH dari pernyataan CREATE TABLE: streamType, alias, dan vertexType. Anda harus menyetel ketiga parameter tersebut untuk tabel aliran. Anda hanya perlu menyetel parameter alias dan vertexType untuk tabel dimensi dan tabel tujuan. streamType: tipe aliran. ETL mengubah aliran menjadi tabel dinamis dan melakukan kueri kontinu pada tabel dinamis untuk menghasilkan tabel dinamis baru. Dalam proses ini, tabel dinamis terus berubah oleh operasi INSERT, UPDATE, dan DELETE. Saat tabel dinamis akhirnya ditulis ke database tujuan, tabel dinamis baru dikonversi kembali menjadi aliran. Saat tabel dinamis baru dikonversi menjadi aliran, Anda harus menentukan parameter ini untuk mengkodekan perubahan tabel dinamis. upsert: aliran upsert. Data dalam tabel dinamis dapat dimodifikasi menggunakan operasi INSERT, UPDATE, dan DELETE. Saat tabel dinamis dikonversi menjadi aliran, operasi INSERT dan UPDATE dikodekan sebagai pesan upsert dan operasi DELETE dikodekan sebagai pesan hapus. Catatan Tabel dinamis yang dikonversi menjadi aliran upsert memerlukan kunci unik. Kunci tersebut bisa komposit. append: aliran append-only. Data dalam tabel dinamis hanya dapat dimodifikasi oleh operasi INSERT. Saat tabel dinamis dikonversi menjadi aliran, hanya baris yang disisipkan yang dikeluarkan. alias: physical table alias yang ditetapkan saat Anda mengonfigurasi database sumber dan tujuan di Langkah 4. vertexType: tipe tabel. Nilai valid: stream: tabel aliran. lookup: tabel dimensi. sink: tabel tujuan.
   Logika Komputasi Transformasi Data	Anda harus mengeksekusi pernyataan CREATE VIEW untuk menggambarkan logika komputasi transformasi data.
   Tabel Tujuan yang Ditransformasi	Anda harus mengeksekusi pernyataan INSERT INTO untuk mendefinisikan tabel tujuan yang ditransformasi.

6. Setelah mengonfigurasi database sumber, database tujuan, dan pernyataan SQL, klik Generate Flink SQL Validation.

   Catatan

   • Anda juga dapat klik Publish untuk memvalidasi pernyataan SQL dan menjalankan pra-pemeriksaan.

   • Jika validasi Flink SQL berhasil, Anda dapat klik untuk melihat detail validasi Flink SQL.

   • Jika validasi Flink SQL gagal, Anda dapat klik untuk melihat detail validasi Flink SQL. Anda dapat mengedit pernyataan SQL berdasarkan pesan kesalahan dan kemudian melakukan validasi Flink SQL lagi.

7. Setelah validasi Flink SQL berhasil, klik Publish untuk menjalankan pra-pemeriksaan.

8. Tunggu hingga tingkat keberhasilan menjadi 100%. Lalu, klik Next: Purchase Instance (Free).

   Catatan

   Jika tugas gagal melewati pra-pemeriksaan, klik View Details di sebelah setiap item yang gagal. Perbaiki masalah berdasarkan penyebabnya dan jalankan pra-pemeriksaan lagi.

9. Di halaman Purchase Instance, pilih Instance Class dan konfirmasi Compute Units (CUs). Nilainya tetap 2 selama pratinjau publik. Lalu, baca dan pilih kotak centang untuk menyetujui Data Transmission Service (Pay-as-you-go) Service Terms dan Service Terms for Public Preview.

   Catatan

   Selama pratinjau publik, setiap pengguna dapat membuat dua instance ETL secara gratis.

10. Klik Buy and Start untuk memulai tugas ETL.