Flink Avro Reader Parquet Writer

flink-parquet-writer.sql

CREATE TABLE kafka_table (
    user_id      STRING,
    order_amount double,
    log_ts       TIMESTAMP(3),
    WATERMARK FOR log_ts AS log_ts - INTERVAL '5' SECOND
) WITH (
  'connector.type'  = 'kafka',
  'connector.topic' = 'users',
  'connector.properties.bootstrap.servers' = 'localhost:9092',
  'connector.properties.group.id' = 'parquet-writer',
  'format.type' = 'avro'
);

CREATE TABLE filesystem_table (
    user_id      STRING,
    order_amount DOUBLE,
    day_ts       STRING,
    hour_ts      STRING
) PARTITIONED BY (day_ts, hour_ts) WITH (
  'connector' = 'filesystem',            
  'path'      = 'hdfs:///path/to/whatever',
  'format'    = 'parquet', 
  'connector.sink.partition-commit.policy' = 'metastore' 
);

INSERT INTO filesystem_table
SELECT user_id, order_amount, DATE_FORMAT(log_ts, 'yyyy-MM-dd'), DATE_FORMAT(log_ts, 'HH')
FROM kafka_table;