功能描述

創(chuàng)建DWS表用于與輸入流連接，從而生成相應的寬表。

前提條件

• 請務必確保您的賬戶下已在數(shù)據(jù)倉庫服務（DWS）里創(chuàng)建了DWS集群。如何創(chuàng)建DWS集群，請參考《數(shù)據(jù)倉庫服務管理指南》中“創(chuàng)建集群”章節(jié)。
• 請確保已創(chuàng)建DWS 數(shù)據(jù)庫 表。
• 該場景作業(yè)需要運行在 DLI 的獨享隊列上，因此要與DWS集群建立增強型跨源連接，且用戶可以根據(jù)實際所需設置相應安全組規(guī)則。
  • 如何建立增強型跨源連接，請參考《 數(shù)據(jù)湖探索 用戶指南》中增強型跨源連接章節(jié)。
  • 如何設置安全組規(guī)則，請參見《 虛擬私有云 用戶指南》中“安全組”章節(jié)。
• Flink跨源開發(fā)場景中直接配置跨源認證信息存在密碼泄露的風險，優(yōu)先推薦您使用DLI提供的跨源認證。

  跨源認證簡介及操作方法請參考跨源認證簡介。

注意事項

創(chuàng)建Flink OpenSource SQL作業(yè)時，在作業(yè)編輯界面的“運行參數(shù)”處，“Flink版本”需要選擇“1.12”，勾選“保存作業(yè)日志”并設置保存作業(yè)日志的 OBS 桶，方便后續(xù)查看作業(yè)日志。

語法格式

 1 2 3 4 5 6 7 8 91011

create table dwsSource (  attr_name attr_type   (',' attr_name attr_type)* )with (  'connector' = 'gaussdb',  'url' = '',  'table-name' = '',  'username' = '',  'password' = '');

參數(shù)說明

示例

從Kafka源表中讀取數(shù)據(jù)，將DWS表作為維表，并將二者生成的寬表信息寫入Kafka結(jié)果表中，其具體步驟如下：

1. 參考增強型跨源連接，在DLI上根據(jù)DWS和Kafka所在的虛擬 私有云 和子網(wǎng)分別創(chuàng)建相應的增強型跨源連接，并綁定所要使用的Flink彈性資源池。
2. 設置DWS和Kafka的安全組，添加入向規(guī)則使其對Flink的隊列網(wǎng)段放通。參考測試地址連通性分別根據(jù)DWS和Kafka的地址測試隊列連通性。若能連通，則表示跨源已經(jīng)綁定成功，否則表示未成功。
3. 連接DWS數(shù)據(jù)庫實例，在DWS中創(chuàng)建相應的表，作為維表，表名為area_info，SQL語句如下：

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

4. 連接DWS數(shù)據(jù)庫實例，向DWS維表area_info中插入測試數(shù)據(jù)，其語句如下：

     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');

5. 參考創(chuàng)建Flink OpenSource作業(yè)，創(chuàng)建flink opensource sql作業(yè)，輸入以下作業(yè)運行腳本，提交運行作業(yè)。該作業(yè)腳本將Kafka作為數(shù)據(jù)源，DWS作為維表，數(shù)據(jù)輸出到Kafka結(jié)果表中。
   注意：創(chuàng)建作業(yè)時，在作業(yè)編輯界面的“運行參數(shù)”處，“Flink版本”選擇“1.12”，勾選“保存作業(yè)日志”并設置保存作業(yè)日志的OBS桶，方便后續(xù)查看作業(yè)日志。如下腳本中的加粗參數(shù)請根據(jù)實際環(huán)境修改。

   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' = 'KafkaSourceTopic',  'properties.bootstrap.servers' = 'KafkaAddress1:KafkaPort,KafkaAddress2:KafkaPort',  'properties.group.id' = 'dws-order',  'scan.startup.mode' = 'latest-offset',  'format' = 'json');--創(chuàng)建地址維表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' = 'gaussdb',  'driver' = 'org.postgresql.Driver',  'url' = 'jdbc:gaussdb://DwsAddress:DwsPort/DwsDbName',  'table-name' = 'area_info',  'username' = 'DwsUserName',  'password' = 'DwsPassword',  'lookup.cache.max-rows' = '10000',  'lookup.cache.ttl' = '2h');--根據(jù)地址維表生成詳細的包含地址的訂單信息寬表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' = 'kafka',  'topic' = 'KafkaSinkTopic',  'properties.bootstrap.servers' = 'KafkaAddress1:KafkaPort,KafkaAddress2:KafkaPort',  'format' = 'json');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;

6. 連接Kafka集群，向kafka中source topic中插入如下測試數(shù)據(jù)：

   {"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"}

7. 連接Kafka集群，讀取kafka中sink topic中數(shù)據(jù)，結(jié)果參考如下：

   {"order_id":"202103241606060001","order_channel":"appShop","order_time":"2021-03-24 16:06:06","pay_amount":200.0,"real_pay":180.0,"pay_time":"2021-03-24 16:10:06","user_id":"0001","user_name":"Alice","area_id":"330106","area_province_name":"a1","area_city_name":"b1","area_county_name":"c2","area_street_name":"d2","region_name":"e1"}{"order_id":"202103251202020001","order_channel":"miniAppShop","order_time":"2021-03-25 12:02:02","pay_amount":60.0,"real_pay":60.0,"pay_time":"2021-03-25 12:03:00","user_id":"0002","user_name":"Bob","area_id":"330110","area_province_name":"a1","area_city_name":"b1","area_county_name":"c4","area_street_name":"d4","region_name":"e1"}{"order_id":"202103251505050001","order_channel":"qqShop","order_time":"2021-03-25 15:05:05","pay_amount":500.0,"real_pay":400.0,"pay_time":"2021-03-25 15:10:00","user_id":"0003","user_name":"Cindy","area_id":"330108","area_province_name":"a1","area_city_name":"b1","area_county_name":"c3","area_street_name":"d3","region_name":"e1"}

常見問題

• Q：若Flink作業(yè)日志中有如下報錯信息，應該怎么解決？

  java.io.IOException: unable to open JDBC writer...Caused by: org.postgresql.util.PSQLException: The connection attempt failed....Caused by: java.net.SocketTimeoutException: connect timed out

  A：應考慮是跨源沒有綁定，或者跨源沒有綁定成功。

  • 參考增強型跨源連接章節(jié)，重新配置跨源。參考DLI跨源連接DWS失敗進行問題排查。

• Q：如果該DWS表在某schema下，則應該如何配置？
  A：如下示例是使用schema為dbuser2下的表area_info：

  --創(chuàng)建地址維表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' = 'gaussdb',  'driver' = 'org.postgresql.Driver',  'url' = 'jdbc:postgresql://DwsAddress:DwsPort/DwsDbname',  'table-name' = 'dbuser2.area_info',  'username' = 'DwsUserName',  'password' = 'DwsPassword',  'lookup.cache.max-rows' = '10000',  'lookup.cache.ttl' = '2h');