?FlinkSQL怎么用

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

目前成都創(chuàng)新互聯(lián)已為超過千家的企業(yè)提供了網(wǎng)站建設(shè)、域名、網(wǎng)站空間、網(wǎng)站托管維護(hù)、企業(yè)網(wǎng)站設(shè)計(jì)、靜海網(wǎng)站維護(hù)等服務(wù)，公司將堅(jiān)持客戶導(dǎo)向、應(yīng)用為本的策略，正道將秉承"和諧、參與、激情"的文化，與客戶和合作伙伴齊心協(xié)力一起成長(zhǎng)，共同發(fā)展。

  
    
  
  
  
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: 主要指計(jì)算列和 watermark
• OPTIONS: WITH (...) 語(yǔ)句內(nèi)定義的 table options
• PARTITIONS: 表分區(qū)信息

• INCLUDING：包含（默認(rèn)行為）
• 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')

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

• 用戶需要指定特性的消費(fèi)時(shí)間戳，即修改 scan.startup.timestamp-millis 屬性
• 用戶想忽略掉解析錯(cuò)誤，需要將 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 語(yǔ)句會(huì)立即執(zhí)行，而 INSERT INTO 語(yǔ)句在調(diào)用 execute 方法時(shí)才會(huì)執(zhí)行
• Table 程序的執(zhí)行入口不夠清晰，像 TableEnvironment.execute() 和 StreamExecutionEnvironment.execute() 都可以觸發(fā) table 程序執(zhí)行
• execute 方法沒有返回值。像 SHOW TABLES 這樣的語(yǔ)句沒有很好地方式返回結(jié)果。另外，sqlUpdate 方法加入了越來越多的語(yǔ)句導(dǎo)致接口定義不清晰，sqlUpdate 可以執(zhí)行 SHOW TABLES 就是一個(gè)反例
• 在 Blink planner 一直提供多 sink 優(yōu)化執(zhí)行的能力，但是在 API 層沒有體現(xiàn)出來

更清晰的執(zhí)行語(yǔ)義

組織多條語(yǔ)句一起執(zhí)行

新舊 API 對(duì)比

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) stored as orc;
create table tbl3 (  m map)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 前綴標(biāo)記 source 和 sink 專有屬性
• format.type 精簡(jiǎn)為 format ，同時(shí) format 自身屬性使用 format 的值作為前綴，比如 csv format 的自身屬性使用 csv 統(tǒng)一作前綴

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

關(guān)于“Flink SQL怎么用”這篇文章就分享到這里了，希望以上內(nèi)容可以對(duì)大家有一定的幫助，使各位可以學(xué)到更多知識(shí)，如果覺得文章不錯(cuò)，請(qǐng)把它分享出去讓更多的人看到。