flink sql实战案例

目录

一、背景

二、流程

三、案例

1.flink sql读取 Kafka 并写入 MySQL

source

sink

insert

2.flinksql读kafka写入kudu

source

sink

insert

四、注意点

1.断点续传

2.实时采集

3.回溯问题

---

一、背景

使用flink sql实时同步一下数据

二、流程

总的来说就三步

source-->>sink->>insert

三、案例

1.flink sql读取 Kafka 并写入 MySQL

source

CREATE TABLE source_table (
    user_id     VARCHAR,
    item_id     VARCHAR,
    category_id VARCHAR,
    behavior    VARCHAR,
    ts          TIMESTAMP
) WITH (
    'connector.type' = 'kafka',           -- 使用 kafka connector
    'connector.version' = 'universal',    -- kafka 版本，universal 支持 0.11 以上的版本
    'connector.topic' = 'user_behavior',  -- kafka topic
    'connector.startup-mode' = 'earliest-offset', -- 从起始 offset 开始读取
    'connector.properties.0.key' = 'zookeeper.connect',  -- 连接信息
    'connector.properties.0.value' = 'localhost:2181', 
    'connector.properties.1.key' = 'bootstrap.servers',
    'connector.properties.1.value' = 'localhost:9092', 
    'update-mode' = 'append',
    'format.type' = 'json',  -- 数据源格式为 json
    'format.derive-schema' = 'true' -- 从 DDL schema 确定 json 解析规则
)

sink

CREATE TABLE sink_table (
    dt VARCHAR,
    pv BIGINT,
    uv BIGINT
) WITH (
    'connector.type' = 'jdbc',     -- 使用 jdbc connector
    'connector.url' = 'jdbc:mysql://localhost:3306/flink-test', -- jdbc url
    'connector.table' = 'pvuv_sink', -- 表名
    'connector.username' = 'username', -- 用户名
    'connector.password' = 'password', -- 密码
    'connector.write.flush.max-rows' = '1' -- 默认5000条，为了演示改为1条
)

insert

INSERT INTO sink_table
SELECT
    DATE_FORMAT(ts, 'yyyy-MM-dd HH:00') as dt,
    COUNT(*)                            as pv,
    COUNT(DISTINCT user_id)             as uv
FROM source_table
GROUP BY DATE_FORMAT(ts, 'yyyy-MM-dd HH:00')

2.flinksql读kafka写入kudu

source

-- kafka source
drop table if exists source_table;
CREATE TABLE source_table (
  user_id VARCHAR
  ,item_id VARCHAR
  ,category_id VARCHAR
  ,behavior INT
  ,ts TIMESTAMP(3)
  ,process_time as proctime()
  , WATERMARK FOR ts AS ts
) WITH (
  'connector' = 'kafka'
  ,'topic' = 'user_behavior'
  ,'properties.bootstrap.servers' = 'venn:9092'
  ,'properties.group.id' = 'source_table'
  ,'scan.startup.mode' = 'group-offsets'
  ,'format' = 'json'
);

sink

-- kafka sink
drop table if exists sink_table;
CREATE TABLE sink_table (
   user_id     STRING
  ,item_id     STRING
  ,category_id STRING
  ,ts          TIMESTAMP(3)
) WITH (
  'connector.type' = 'kudu'
  ,'kudu.masters' = 'venn:7051,venn:7151,venn:7251'
  ,'kudu.table' = 'source_table'
  ,'kudu.hash-columns' = 'user_id'
  ,'kudu.primary-key-columns' = 'user_id'
  ,'kudu.max-buffer-size' = '5000'
  ,'kudu.flush-interval' = '1000'
);

insert

-- insert
insert into sink_table
select user_id, item_id, category_id,ts
from source_table;

四、注意点

1.断点续传

断点续传是指数据同步任务在运行过程中因各种原因导致任务失败，不需要重头同步数据，只需要从上次失败的位置继续同步即可，类似于下载文件时因网络原因失败，不需要重新下载文件，只需要继续下载就行，可以大大节省时间和计算资源。

默认关闭，开启的话调参 isRestore：true

 

2.实时采集

根据数据源的数据是否实时变化可以把数据同步分为离线数据同步和实时数据同步，上面介绍的断点续传就是离线数据同步里的功能，实时采集其实就是实时数据同步，当数据源里的数据发生了增删改操作，同步任务监听到这些变化，将变化的数据实时同步到目标数据源。除了数据实时变化外，实时采集和离线数据同步的另一个区别是：实时采集任务是不会停止的，任务会一直监听数据源是否有变化。

3.回溯问题

例如mysql是事务型数据库，会update，最新的消息发过去，得回撤更新前的消息，update-和update+两条消息，数据都在state里。

简单举个例子，统计男女数量，一开始mysql里是男，然后mysql更新为女了，这时候你接收的kafka，消息都会过来，state里一开始存着男，然后把男回撤，女进来，就要删除男新增女，state一般在rocksdb里，可以设置table.exec.state.ttl 窗口时间。

相关参数
val tEnv: TableEnvironment = ...
val configuration = tEnv.getConfig().getConfiguration()
 
configuration.setString("table.exec.mini-batch.enabled", "true")         // 启用
configuration.setString("table.exec.mini-batch.allow-latency", "5 s")    // 缓存超时时长
configuration.setString("table.exec.mini-batch.size", "5000")            // 缓存大小

 

 

ps:因为本人这方面不是很专业，还在学习的阶段，有问题的话大家可以多多指教哈～