FlinkSQL 平台

CREATE TEMPORARY FUNCTION MillisecondsToDateStr AS 'io.github.shengjk.udf.MillisecondsToDateStr' LANGUAGE JAVA;
-- ExecutionCheckpointingOptions
set execution.checkpointing.mode=EXACTLY_ONCE;
set execution.checkpointing.timeout=30 min;--  30min
set execution.checkpointing.interval=1 min ; -- 1min
set execution.checkpointing.externalized-checkpoint-retention=RETAINONCANCELLATION;
-- ExecutionConfigOptions
set table.exec.state.ttl=1 day;  -- 1 day
set table.exec.mini-batch.enabled=true; -- enable mini-batch optimization
set table.exec.mini-batch.allow-latency=1 s; -- 1s
set table.exec.mini-batch.size=1000;
set table.exec.sink.not-null-enforcer=drop;
-- -- dadadadadada
CREATE TABLE orders
(
    status      int,
    courier_id  bigint,
    id          bigint,
    finish_time BIGINT
)
WITH (
    'connector' = 'kafka','topic' = 'canalmonitororder',
    'properties.bootstrap.servers' = 'localhost:9092','properties.group.id' = 'testGroup',
    'format' = 'ss-canal-json','ss-canal-json.table.include' = 'orders','scan.startup.mode' = 'earliest-offset');
-- flink.partition-discovery.interval-millis;
CREATE TABLE infos
(
    info_index int,
    order_id   bigint
)
WITH (
    'connector' = 'kafka','topic' = 'canalmonitororder',
    'properties.bootstrap.servers' = 'localhost:9092','properties.group.id' = 'testGroup',
    'format' = 'ss-canal-json','ss-canal-json.table.include' = 'infos','scan.startup.mode' = 'earliest-offset');
CREATE TABLE redisCache
(
    finishOrders BIGINT,
    courier_id   BIGINT,
    dayStr       String
)
WITH (
    'connector' = 'redis',
    'hostPort'='localhost:6400',
    'keyType'='hash',
    'keyTemplate'='test2${courierid}',
    'fieldTemplate'='${dayStr}',
    'valueNames'='finishOrders',
    'expireTime'='259200');
create view temp as
select o.courier_id,
       (CASE
            WHEN sum(infosMaxIndex.info_index) is null then 0
            else sum(infosMaxIndex.info_index) end) finishOrders,
       o.status,
       dayStr
from ((select courier_id,
              id,
              last_value(status)                             status,
              MillisecondsToDateStr(finish_time, 'yyyyMMdd') dayStr
       from orders
       where status = 60
       group by courierid, id, MillisecondsToDateStr(finishtime, 'yyyyMMdd'))) o
         left join (select max(infoindex) infoindex, order_id
                    from infos
                    group by orderid) infosMaxIndex on o.id = infosMaxIndex.orderid
group by o.courier_id, o.status, dayStr;
INSERT INTO redisCache SELECT finishOrders,courier_id,dayStr FROM temp;

flink-1.12.0/bin/flink  run -p 3 -yt ./flinkjar/  -C file:///home/shengjk/flinkjar/test-udf.jar -C file:///home/shengjk/flinkjar/jedis-2.10.2.jar  -m yarn-cluster -ynm sqlDemo  -c io.github.shengjk.Main ./flinksql-platform-1.0-SNAPSHOT.jar --sqlPath ./xxx.sql