本文主要是介绍Flink 流转表,表转流,watermark设置,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!
流转表
首先创建一个流
@Data
@AllArgsConstructor
@NoArgsConstructor
public static class Nan {private String xing;private String name;private Long ts;
}StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
StreamTableEnvironment tenv = StreamTableEnvironment.create(env);DataStreamSource<String> sourceNan = env.socketTextStream("hdp01", 1111);
DataStreamSource<String> sourceNv = env.socketTextStream("hdp01", 2222);System.setProperty("java.net.preferIPv4Stack", "true");SingleOutputStreamOperator<Nan> beanNan = sourceNan.map(new MapFunction<String, Nan>() {@Overridepublic Nan map(String s) throws Exception {try {String[] split = s.split(",");return new Nan(split[0].substring(0, 1), split[1], Long.parseLong(split[2]));} catch (Exception e) {return null;}}
}).filter(Objects::nonNull).assignTimestampsAndWatermarks(WatermarkStrategy.<Nan>forMonotonousTimestamps().withTimestampAssigner(new SerializableTimestampAssigner<Nan>() {@Overridepublic long extractTimestamp(Nan nan, long l) {return nan.getTs();}
})).returns(TypeInformation.of(Nan.class));
创建watermark
流转表的时候有一个点要注意,watermark必须要重新指定,否则会丢失,常用的方式如下
创建watermark,有两步,
第一步:必须要依据一个字段来创建watermark,这个字段必须是timestamp_ltz(3)的类型。
第二步:根据时间戳字段生成watermark
时间戳字段有两种获取方式
1、根据一个bigint字段进行转换
2、在流转表,且流上设置了watermark的情况下,根据内置属性rowtime创建,这个rowtime是流转表时暴露出来的事件时间
watermark也有两种获取方式
1、根据时间戳字段重新创建watermark
2、在流转表,且流上设置了watermark的情况下,沿用流上的watermark
下面是两种场景,只要记住第一种就行了,其实第二种没什么用。
1、 根据一个bigint字段进行创建时间戳字段,然后重新创建watermark
tenv.createTemporaryView("nan", beanNan, Schema.newBuilder().column("xing", DataTypes.STRING()).column("name", DataTypes.STRING()).column("ts", DataTypes.BIGINT()).columnByExpression("rt", "to_timestamp_ltz(ts,3)") // 根据一个bigint字段进行转换.watermark("rt", "rt - interval '1' second ") // 重新创建watermark.build());
2、根据内置属性rowtime创建时间戳字段,然后沿用流上的watermark
tenv.createTemporaryView("nan1", beanNan, Schema.newBuilder().column("xing", DataTypes.STRING()).column("name", DataTypes.STRING()).column("ts", DataTypes.BIGINT()).columnByMetadata("rt", DataTypes.TIMESTAMP_LTZ(3),"rowtime") // 根据内置属性rowtime创建.watermark("rt", "source_watermark()") // 沿用流的watermark “source_watermark 等于 rt - interval '1' second”.build());
TableResult tableResult = tenv.executeSql("select *,current_watermark(rt) from nan");
tableResult.print();
表转流
首先创建一个表
String source = "CREATE TABLE person ( " +" xing STRING, " +" name STRING, " +" ts BIGINT, " +" rt as to_timestamp_ltz(ts,3), " +" watermark for rt as rt - interval '1' second " +") WITH ( " +" 'connector' = 'kafka', " +" 'topic' = 'flink_topic', " +" 'properties.bootstrap.servers' = '172.16.10.139:9092', " +" 'properties.group.id' = 'testGroup', " +" 'scan.startup.mode' = 'latest-offset', " +" 'format' = 'json' " +")";tenv.executeSql(source);
创建watermark
表转流,可以沿用流上的watermark,不需要额外声明
DataStream<Row> dataStream = tenv.toDataStream(table);dataStream.process(new ProcessFunction<Row, Object>() {@Overridepublic void processElement(Row value, ProcessFunction<Row, Object>.Context ctx, Collector<Object> out) throws Exception {System.out.println(value+" watermark=>"+ctx.timerService().currentWatermark());}
});
env.execute();
这篇关于Flink 流转表,表转流,watermark设置的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!