Data Transmission Service:Konfigurasikan tugas ETL dalam mode Flink SQL

Parameter dalam klausa WITH

Setiap pernyataan CREATE TABLE menggunakan klausa WITH untuk mengonfigurasi perilaku ETL tabel tersebut:

Contoh: join stream table dengan dimension table

Skrip berikut melakukan join stream table test_orders dengan dimension table product dan memasukkan hasilnya ke tabel tujuan test_orders_new.

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`;