flink sql中的自定义sink connector如何获取到source table中定义的event time和watermark?

Tue, 20 Feb 2024 00:35:47 -0800 

请问flink sql中的自定义sink connector如何获取到source table中定义的event time和watermark?


public class XxxSinkFunction extends RichSinkFunction<RowData> implements 
CheckpointedFunction, CheckpointListener {


    @Override
    public synchronized void invoke(RowData rowData, Context context) throws 
IOException {
           //  这里想从rowData中获取event time和watermark值，如何实现呢？
    }
}


例如source table如下定义


    CREATE TEMPORARY TABLE source_table(
  username varchar,
  click_url varchar,
  eventtime varchar,                                                        
  ts AS TO_TIMESTAMP(eventtime),
  WATERMARK FOR ts AS ts -INTERVAL'2'SECOND--为Rowtime定义Watermark。
) with (
  'connector'='kafka',
  ...

);


CREATE TEMPORARY TABLE sink_table(
  username varchar,
  click_url varchar,
  eventtime varchar
) with (
  'connector'='xxx',
  ...
);
insert into sink_table select username,click_url,eventtime from source_table;

回复