揭秘｜字节跳动基于Flink SQL的流式数据质量监控（下）实践细节

-- SOURCE TABLE DDL：定义了Kafka数据源及消费策略
create table data_source (
    app_id BIGINT,
    name VARCHAR,
    _meta_timestamp BIGINT,
    time_formatted VARCHAR,
    price DOUBLE,
    type VARCHAR
) with (
    'format.type' = 'json',
    'update-mode' = 'append',
    'metadata.fields.mapping' = 'timestamp=_meta_timestamp',
    'connector.version' = '0.10',
    'format.derive-schema' = 'true',
    'connector.topic' = 'kafka_topic',
    'connector.group.id' = 'kafka_group_id',
    'connector.startup-mode' = 'latest-offset',
    'connector.type' = 'kafka',
    'format.skip-dirty' = 'true',
    ...
);


-- SINK TABLE DDL：定义了TSDB Sink
create table data_sink(
    type varchar,
    name varchar,
    val double,
    tags varchar
) with (
    'connector.type' = 'tsdb',
    'connector.tsdb.table' = 'tsdb_table',
    ...
);


-- RULE SQL 1：计算字段app_id的空值率
INSERT INTO data_sink 
SELECT 
    'COUNTER' AS type, 
    'r0' AS name, 
    CAST(1.0 AS DOUBLE) AS val, 
    CONCAT('{', IF(`app_id` is null,  'app_id=nit', 'app_id=normal') , ',' , IF(`name` is null,  'name=nit', 'name=normal') ,'}') AS tags 
FROM data_source;


-- RULE SQL 2：以time_formatted为事件时间，计算数据写入延迟
INSERT INTO data_sink 
SELECT 
    'TIMER' AS type, 
    'r1' AS name, 
    CAST((_meta_timestamp - UNIX_TIMESTAMP(time_formatted, 'yyyy-MM-dd HH:mm:ss') * 1000) AS DOUBLE) AS val, 
    '{}' AS tags 
FROM data_source;


-- RULE SQL 3：计算指标app_id的取值范围
INSERT INTO data_sink 
SELECT 
    'TIMER' AS type, 
    'r2.app_id' AS name, 
    CAST(app_id AS DOUBLE) AS val, 
    '{}' AS tags 
FROM data_source 
WHERE app_id IS NOT NULL;