@Jane Chan

SET 'execution.checkpointing.interval' = '10s';

CREATE TABLE kafka_log (
  host VARCHAR(255),
  phone VARCHAR(255),
  idname VARCHAR(255),
  idvalue VARCHAR(255),
  time_stamp AS cast(CURRENT_TIMESTAMP as timestamp(3)),
  WATERMARK FOR time_stamp AS time_stamp - INTERVAL '5' SECOND
) WITH (
  'connector' = 'kafka',
  'topic' = 'test11',
  'properties.bootstrap.servers' = 'localhost:9092',
  'properties.zookeeper.connect' = 'localhost:2181',
  'properties.group.id' = 'flink-runtime-001',
  'scan.startup.mode' = 'latest-offset',
  'format' = 'csv',
  'csv.field-delimiter' = ','
);

CREATE TABLE fileOut( 
                host string, 
                idname string, 
                idvalue string
                ) WITH ( 
                'connector' = 'filesystem', 
                'path' = '/opt/code/nd/log',
                'format' = 'csv' 
                );


INSERT INTO fileOut
SELECT host, idname, idvalue
FROM (
  SELECT host, idname, idvalue, 
         COUNT(DISTINCT phone) AS phone_count, 
  FROM kafka_log
  GROUP BY host, idname, idvalue,
           TUMBLE(time_stamp, INTERVAL '1' MINUTE)
) AS t
WHERE phone_count = 1;

This is a simple window processing process, which is processed once a minute. Is writing result into csv file also written once a minute? The situation I encountered was that I waited for a long time before the csv file was written successfully. I didn't know how to locate and debug this kind of window function.