Dimension Tables

Syntax

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

create table dwsSource (
  attr_name attr_type 
  (',' attr_name attr_type)* 
)
with (
  'connector' = 'dws',
  'url' = '',
  'tableName' = '',
  'username' = '',
  'password' = ''
);

Parameter Description

Examples

Read data from a Kafka source table, use a DWS table as the dimension table. Write wide table information generated by the source and dimension tables to a Print result table. The procedure is as follows:

1. Connect to the DWS database instance, create a table as a dimension table, and name the table area_info. Example SQL statements are as follows:

   1 2 3 4 5 6 7 8 9

   create table public.area_info( area_id VARCHAR, area_province_name VARCHAR, area_city_name VARCHAR, area_county_name VARCHAR, area_street_name VARCHAR, region_name VARCHAR, PRIMARY KEY(area_id) );

2. Connect to the database and run the following statement to insert test data into the dimension table area_info:

   1 2 3 4 5 6 7

   insert into area_info (area_id, area_province_name, area_city_name, area_county_name, area_street_name, region_name) values ('330102', 'a1', 'b1', 'c1', 'd1', 'e1'), ('330106', 'a1', 'b1', 'c2', 'd2', 'e1'), ('330108', 'a1', 'b1', 'c3', 'd3', 'e1'), ('330110', 'a1', 'b1', 'c4', 'd4', 'e1');

3. Run the flink sql command to create the source table, result table, and dimension table and run the following SQL statements:

   CREATE TABLE orders (
     order_id string,
     order_channel string,
     order_time string,
     pay_amount double,
     real_pay double,
     pay_time string,
     user_id string,
     user_name string,
     area_id string,
     proctime as Proctime()
   ) WITH (
     'connector' = 'kafka',
     'topic' = 'order_test',
     'properties.bootstrap.servers' = 'KafkaAddress1:KafkaPort,KafkaAddress2:KafkaPort',
     'properties.group.id' = 'dws-order',
     'scan.startup.mode' = 'latest-offset',
     'format' = 'json'
   );
   --Create the address dimension table.
   create table area_info (
       area_id string, 
       area_province_name string, 
       area_city_name string, 
       area_county_name string,
       area_street_name string, 
       region_name string 
   ) WITH (
     'connector' = 'dws',
     'url' = 'jdbc:gaussdb://DwsAddress:DwsPort/DwsDbName',
     'tableName' = 'area_info',
     'username' = 'DwsUserName',
     'password' = 'password',
     'lookupCacheMaxRows' = '10000',
     'lookupCacheExpireAfterAccess' = '2h'
   );
   --Create a wide table based on the address dimension table for detailed order information.
   create table order_detail(
       order_id string,
       order_channel string,
       order_time string,
       pay_amount double,
       real_pay double,
       pay_time string,
       user_id string,
       user_name string,
       area_id string,
       area_province_name string,
       area_city_name string,
       area_county_name string,
       area_street_name string,
       region_name string
   ) with (
     'connector' = 'print'
   
   );
   insert into order_detail
       select orders.order_id, orders.order_channel, orders.order_time, orders.pay_amount, orders.real_pay, orders.pay_time, orders.user_id, orders.user_name,
              area.area_id, area.area_province_name, area.area_city_name, area.area_county_name,
              area.area_street_name, area.region_name  from orders
       left join area_info for system_time as of orders.proctime as area on orders.area_id = area.area_id;

4. Write data to Kafka.

   1 2 3

   {"order_id":"202103241606060001", "order_channel":"appShop", "order_time":"2021-03-24 16:06:06", "pay_amount":"200.00", "real_pay":"180.00", "pay_time":"2021-03-24 16:10:06", "user_id":"0001", "user_name":"Alice", "area_id":"330106"} {"order_id":"202103251202020001", "order_channel":"miniAppShop", "order_time":"2021-03-25 12:02:02", "pay_amount":"60.00", "real_pay":"60.00", "pay_time":"2021-03-25 12:03:00", "user_id":"0002", "user_name":"Bob", "area_id":"330110"} {"order_id":"202103251505050001", "order_channel":"qqShop", "order_time":"2021-03-25 15:05:05", "pay_amount":"500.00", "real_pay":"400.00", "pay_time":"2021-03-25 15:10:00", "user_id":"0003", "user_name":"Cindy", "area_id":"330108"}

5. The following figure shows the result.

   

FAQs

• Q: How do I check whether dimension tables support pushdown?

  A: By default, Flink enables condition pushdown. In addition, dws-connector 2.1.0.1 and later versions support condition pushdown of dimension tables.

  Identification method: Check the details of the lookup join operator on the Flink UI to see if there is a WHERE condition. Check the number of records displayed in Records Received and Records Sent to determine whether pushdown has taken effect.

• Q: Why does pushdown not take effect even though a WHERE condition is specified?

  A: Although Flink enables condition pushdown by default, pushdown also depends on the capabilities of the data source. there are still many cases where Flink cannot perform condition pushdown:

  • Case 1: Check whether 'lookup.filter-push-down.enabled' = 'false' is configured in Flink. If yes, the condition pushdown is disabled.
  • Case 2: The operators in Flink SQL contain complex functions or expressions, such as JSON parsing, array operations, UDFs, regular expressions, or non-deterministic functions (such as RAND()).
  • Case 3: If the filter criteria in Flink SQL are metadata content or implicit time fields, pushdown is generally not supported. These fields are filled only after the operator execution is complete in physical storage. The source end cannot filter data based on these fields during operator execution.
  • Case 4: If the field type in Flink SQL is different from that in the data source, Flink performs implicit conversion. As a result, the pushdown fails.

    For example, if the field in the data source is of the SMALLINT type and the field in Flink SQL is of the INTEGER type, Flink optimizes the field to CAST (column AS INTEGER = XXX) and the pushdown fails.