说明

本博客周五更新一次
Flink Sql 支持众多连接器，语句各有不同，使用时查找起来麻烦，找到了也可能是错的，因此我整理收集了已知的连接与窗口实例，并持续更新，在此分享出来。

大数据博客列表

资料

datagen文档
jdbc
kafka
hive
- 基础文档不同版本hive使用不同jar
- hive读写文档
hbase文档

实例

连接器

随机数连接器

生成随机数表 datagen ，每5秒生成一次

CREATE TABLE source_table (
  f0 INT,
  f1 INT,
  f2 STRING
 ) WITH (
  'connector' = 'datagen',
  'rows-per-second'='5'
 );

输出连接器

CREATE TABLE print_table (
  f0 INT,
  f1 INT,
  f2 STRING
) WITH (
  'connector' = 'print'
);

kafka连接器

CREATE TABLE input_kafka (
  `user_id` BIGINT,
  `page_id` BIGINT,
  `status` STRING 
) WITH (
  'connector' = 'kafka',
  'topic' = 'input_kafka',
  'properties.bootstrap.servers' = 'localhost:9092',
  'properties.group.id' = 'testGroup',
  'scan.startup.mode' = 'latest-offset',
  'format' = 'json',
  --以下非必填
  'json.fail-on-missing-field' = 'false',
  'json.ignore-parse-errors' = 'true',
  'properties.zookeeper.connect' = '172.25.20.76:2181/kafka'
)

hive连接器

hive表必须要开启checkpoint

CREATE CATALOG testmyhive WITH (
    'type' = 'hive',
    'default-database' = 'zhp',
    'hive-conf-dir' = '/Users/huipeizhu/hive-conf'
);

USE CATALOG testmyhive;

--- 清空表，防止多余
drop table IF EXISTS item_test;

drop table IF EXISTS hive_flink_table;

---- 创建kafka表
create table item_test ( 
     itemId BIGINT,
     price BIGINT,
     proctime AS PROCTIME ()
 )with ( 
     'connector' = 'kafka',
     'topic' = 'flink-catalog-v1',  
     'properties.bootstrap.servers'='127.0.0.1:9092',
     'properties.group.id'='test-1',
     'format'='json',
     'scan.startup.mode' = 'earliest-offset'
 );

--- 创建hive表
SET table.sql-dialect=hive;

CREATE TABLE hive_flink_table (
 itemId BIGINT, 
 price BIGINT, 
 ups string
) TBLPROPERTIES (
  'sink.rolling-policy.rollover-interval'='1min',
  'sink.partition-commit.trigger'='process-time',
  'sink.partition-commit.policy.kind'='metastore,success-file'
);

SET table.sql-dialect=default;

insert into hive_flink_table select itemId,price, 'XXXXaaa' as ups from item_test;

mysql连接器

CREATE TABLE sync_test_1 (
  day_time string,
  total_gmv bigint,
  PRIMARY KEY (day_time) NOT ENFORCED
 ) WITH (
   'connector' = 'jdbc',
   'url' = 'jdbc:mysql://172.25.21.10:3306/flink_web?characterEncoding=UTF-8',
   'table-name' = 'sync_test_1',
   'username' = 'videoweb',
   'password' = 'suntek'
 );

Elasticsearch连接器

CREATE TABLE enriched_orders (
  order_id INT,
  order_date TIMESTAMP(0),
  customer_name STRING,
  price DECIMAL(10, 5),
  product_id INT,
  order_status BOOLEAN,
  product_name STRING,
  product_description STRING,
  shipment_id INT,
  origin STRING,
  destination STRING,
  is_arrived BOOLEAN,
  PRIMARY KEY (order_id) NOT ENFORCED
) WITH (
    'connector' = 'elasticsearch-7',
    'hosts' = 'http://172.25.23.15:9401',
    'index' = 'enriched_orders'
);

hbase

CREATE TABLE hTable (
 rowkey INT,
 family1 ROW<q1 INT>,
 family2 ROW<q2 STRING, q3 BIGINT>,
 family3 ROW<q4 DOUBLE, q5 BOOLEAN, q6 STRING>,
 PRIMARY KEY (rowkey) NOT ENFORCED
) WITH (
  --- hbase 1.4.x 对应1.4   2.2.x 对应 2.2
 'connector' = 'hbase-1.4',
 'table-name' = 'mytable',
 'zookeeper.quorum' = 'localhost:2181'
);

窗口

滚动窗口

滚动窗口表，其中 createTime 为自定义时间戳字段，每5秒执行一次。

select 
  userId,count(*) as orderCount,max(money) as maxMoney,min(money) as minMoney 
  from t_order 
group by 
  userId,
  tumble(createTime,Interval '5' SECOND)