?Flink SQL怎么用

這篇文章將為大家詳細講解有關Flink SQL怎么用，小編覺得挺實用的，因此分享給大家做個參考，希望大家閱讀完這篇文章后可以有所收獲。

  
    
  
  
  
CREATE [TEMPORARY] TABLE base_table (    id BIGINT,    name STRING,    tstmp TIMESTAMP,    PRIMARY KEY(id)) WITH (    'connector': 'kafka') CREATE [TEMPORARY] TABLE derived_table (    WATERMARK FOR tstmp AS tsmp - INTERVAL '5' SECOND)LIKE base_table;
  
    
  
  
  


  
    
  
  
  

  
    
  
  
  
CREATE [TEMPORARY] TABLE derived_table (    id BIGINT,    name STRING,    tstmp TIMESTAMP,    PRIMARY KEY(id),    WATERMARK FOR tstmp AS tsmp - INTERVAL '5' SECOND) WITH (    ‘connector’: ‘kafka’)
  
    
  
  
  
  
     
   
   
   

  
    
  
  
  

多屬性策略

• ALL：完整的表定義
• CONSTRAINTS: primary keys, unique key 等約束
• GENERATED: 主要指計算列和 watermark
• OPTIONS: WITH (...) 語句內定義的 table options
• PARTITIONS: 表分區信息

• INCLUDING：包含（默認行為）
• EXCLUDING：排除
• OVERWRITING：覆蓋

  
    
  
  
  CREATE [TEMPORARY] TABLE base_table (    id BIGINT,    name STRING,    tstmp TIMESTAMP,    PRIMARY KEY(id)) WITH (    'connector': 'kafka',    'scan.startup.specific-offsets': 'partition:0,offset:42;partition:1,offset:300',    'format': 'json') CREATE [TEMPORARY] TABLE derived_table (    WATERMARK FOR tstmp AS tsmp - INTERVAL '5' SECOND)WITH (    'connector.starting-offset': '0')LIKE base_table (OVERWRITING OPTIONS, EXCLUDING CONSTRAINTS);

  
    
  
  
  
CREATE [TEMPORARY] TABLE derived_table (    id BIGINT,    name STRING,    tstmp TIMESTAMP,    WATERMARK FOR tstmp AS tsmp - INTERVAL '5' SECOND) WITH (    'connector': 'kafka',    'scan.startup.specific-offsets': 'partition:0,offset:42;partition:1,offset:300',    'format': 'json')
  
    
  
  
  
  
     
   
   
   

  
    
  
  
  

create table kafka_table (  id bigint,  age int,  name STRING) WITH (  'connector' = 'kafka',  'topic' = 'employees',  'scan.startup.mode' = 'timestamp',  'scan.startup.timestamp-millis' = '123456',  'format' = 'csv',  'csv.ignore-parse-errors' = 'false')

在之前的版本，如果用戶有如下需求：

• 用戶需要指定特性的消費時間戳，即修改 scan.startup.timestamp-millis 屬性
• 用戶想忽略掉解析錯誤，需要將 format.ignore-parse-errors 改為 true

  
    
  
  
  
table_name /*+ OPTIONS('k1'='v1', 'aa.bb.cc'='v2') */
  
    
  
  
  
  
     
   
   
   

  
    
  
  
  

  
    
  
  
  
CREATE TABLE kafka_table1 (id BIGINT, name STRING, age INT) WITH (...);CREATE TABLE kafka_table2 (id BIGINT, name STRING, age INT) WITH (...);
-- override table options in query sourceselect id, name from kafka_table1 /*+ OPTIONS('scan.startup.mode'='earliest-offset') */;
-- override table options in joinselect * from    kafka_table1 /*+ OPTIONS('scan.startup.mode'='earliest-offset') */ t1    join    kafka_table2 /*+ OPTIONS('scan.startup.mode'='earliest-offset') */ t2    on t1.id = t2.id;
-- override table options for INSERT target tableinsert into kafka_table1 /*+ OPTIONS('sink.partitioner'='round-robin') */ select * from kafka_table2;
  
    
  
  
  


  
    
  
  
  

  
    
  
  
  
// instantiate table environmentTableEnvironment tEnv = ...// access flink configurationConfiguration configuration = tEnv.getConfig().getConfiguration();// set low-level key-value optionsconfiguration.setString("table.dynamic-table-options.enabled", "true");
  
    
  
  
  


  
    
  
  
  

• 原先的 sqlUpdate() 方法傳遞 DDL 語句會立即執行，而 INSERT INTO 語句在調用 execute 方法時才會執行
• Table 程序的執行入口不夠清晰，像 TableEnvironment.execute() 和 StreamExecutionEnvironment.execute() 都可以觸發 table 程序執行
• execute 方法沒有返回值。像 SHOW TABLES 這樣的語句沒有很好地方式返回結果。另外，sqlUpdate 方法加入了越來越多的語句導致接口定義不清晰，sqlUpdate 可以執行 SHOW TABLES 就是一個反例
• 在 Blink planner 一直提供多 sink 優化執行的能力，但是在 API 層沒有體現出來

更清晰的執行語義

組織多條語句一起執行

新舊 API 對比

execute vs createStatementSet

  
    
  
  
  
EnvironmentSettings settings = EnvironmentSettings.newInstance()...build();TableEnvironment tableEnv = TableEnvironment.create(settings);// to use hive dialecttableEnv.getConfig().setSqlDialect(SqlDialect.HIVE);// use the hive catalogtableEnv.registerCatalog(hiveCatalog.getName(), hiveCatalog);tableEnv.useCatalog(hiveCatalog.getName());
  
    
  
  
  


  
    
  
  
  

  
    
  
  
  
create external table tbl1 (  d decimal(10,0),  ts timestamp)partitioned by (p string)location '%s'tblproperties('k1'='v1');  create table tbl2 (s struct<ts:timestamp,bin:binary>) stored as orc;
create table tbl3 (  m map<timestamp,binary>)partitioned by (p1 bigint, p2 tinyint)row format serde 'org.apache.hadoop.hive.serde2.lazybinary.LazyBinarySerDe';
create table tbl4 (  x int,  y smallint)row format delimited fields terminated by '|' lines terminated by '\n';
  
    
  
  
  


  
    
  
  
  

• 使用 connector 作為 connector 的類型 key，connector 版本信息直接放到 value 中，比如 0.11 的 kafka 為 kafka-0.11 
• 去掉了其余屬性中多余的 connector 前綴
• 使用 scan 和 sink 前綴標記 source 和 sink 專有屬性
• format.type 精簡為 format ，同時 format 自身屬性使用 format 的值作為前綴，比如 csv format 的自身屬性使用 csv 統一作前綴

  
    
  
  
  CREATE TABLE kafkaTable ( user_id BIGINT, item_id BIGINT, category_id BIGINT, behavior STRING, ts TIMESTAMP(3)) WITH ( 'connector' = 'kafka', 'topic' = 'user_behavior', 'properties.bootstrap.servers' = 'localhost:9092', 'properties.group.id' = 'testGroup', 'format' = 'csv', 'scan.startup.mode' = 'earliest-offset')

CREATE CATALOG mypg WITH(    'type' = 'jdbc',    'default-database' = '...',    'username' = '...',    'password' = '...',    'base-url' = '...');
USE CATALOG mypg;

DDL 定義 python UDF

  
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();BatchTableEnvironment tEnv = BatchTableEnvironment.create(env);
tEnv.getConfig().getConfiguration().setString("python.files", "/home/my/test1.py");tEnv.getConfig().getConfiguration().setString("python.client.executable", "python3");tEnv.sqlUpdate("create temporary system function func1 as 'test1.func1' language python");Table table = tEnv.fromDataSet(env.fromElements("1", "2", "3")).as("str").select("func1(str)");
tEnv.toDataSet(table, String.class).collect();
 
  

向量化支持

  
@udf(input_types=[DataTypes.BIGINT(), DataTypes.BIGINT()], result_type=DataTypes.BIGINT(), udf_type="pandas")def add(i, j):  return i + j
table_env = BatchTableEnvironment.create(env)
# register the vectorized Python scalar functiontable_env.register_function("add", add)
# use the vectorized Python scalar function in Python Table APImy_table.select("add(bigint, bigint)")
# use the vectorized Python scalar function in SQL APItable_env.sql_query("SELECT add(bigint, bigint) FROM MyTable")

關于“Flink SQL怎么用”這篇文章就分享到這里了，希望以上內容可以對大家有一定的幫助，使各位可以學到更多知識，如果覺得文章不錯，請把它分享出去讓更多的人看到。