前提條件

請務必確保您的賬戶下已創(chuàng)建了相應實例。

注意事項

• 創(chuàng)建Flink OpenSource SQL作業(yè)時，在作業(yè)編輯界面的“運行參數(shù)”處，“Flink版本”需要選擇“1.12”，勾選“保存作業(yè)日志”并設置保存作業(yè)日志的 OBS 桶，方便后續(xù)查看作業(yè)日志。
• Flink跨源開發(fā)場景中直接配置跨源認證信息存在密碼泄露的風險，優(yōu)先推薦您使用 DLI 提供的跨源認證。

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

語法格式

 1 2 3 4 5 6 7 8 9101112

CREATE TABLE  table_id (  attr_name attr_type   (',' attr_name attr_type)* )  WITH (  'connector' = 'jdbc',  'url' = '',  'table-name' = '',  'driver' = '',  'username' = '',  'password' = '');

參數(shù)說明

數(shù)據(jù)類型映射

示例

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

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

   CREATE TABLE `flink`.`area_info` (`area_id` VARCHAR(32) NOT NULL,`area_province_name` VARCHAR(32) NOT NULL,`area_city_name` VARCHAR(32) NOT NULL,`area_county_name` VARCHAR(32) NOT NULL,`area_street_name` VARCHAR(32) NOT NULL,`region_name` VARCHAR(32) NOT NULL,PRIMARY KEY (`area_id`))ENGINE = InnoDBDEFAULT CHARACTER SET = utf8mb4COLLATE = utf8mb4_general_ci;

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

   insert into flink.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ù)源，JDBC作為維表，數(shù)據(jù)寫入到Kafka結果表。
   注意：創(chuàng)建作業(yè)時，在作業(yè)編輯界面的“運行參數(shù)”處，“Flink版本”選擇“1.12”，勾選“保存作業(yè)日志”并設置保存作業(yè)日志的OBS桶，方便后續(xù)查看作業(yè)日志。如下腳本中的加粗參數(shù)請根據(jù)實際環(huán)境修改。

   1 2 3 4 5 6 7 8 910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364

   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' = 'jdbc-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' = 'jdbc', 'url' = 'jdbc:mysql://JDBCAddress:JDBCPort/flink',--其中url中的flink表示MySQL中area_info表所在的數(shù)據(jù)庫名 'table-name' = 'area_info', 'username' = 'JDBCUserName', 'password' = 'JDBCPassWord');--根據(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ù)，結果參考如下：

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

常見問題

無。