1 Star 10 Fork 16

栾昊/Flink_1.13

加入 Gitee
与超过 1200万 开发者一起发现、参与优秀开源项目,私有仓库也完全免费 :)
免费加入
文件
.idea
input
src/main
java/com/hao
chapter05
chapter06
chapter07
chapter08
BillCheckExample.java
CoGroupTest.java
ConnectTest.java
IntervalJoinTest.java
SplitStreamTest.java
UnionTest.java
WindowJoinTest.java
chapter09
chapter11
chapter12
wc
resources
pom.xml
该仓库未声明开源许可证文件(LICENSE),使用请关注具体项目描述及其代码上游依赖。
克隆/下载
UnionTest.java 2.65 KB
一键复制 编辑 原始数据 按行查看 历史
dahuanhuan 提交于 3年前 . Union合流操作案例
package com.hao.chapter08;
import com.hao.chapter05.Event;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import java.time.Duration;
public class UnionTest {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
DataStreamSource<String> streamSource1 = env.socketTextStream("hadoop102", 8888);
DataStreamSource<String> streamSource2 = env.socketTextStream("hadoop103", 8888);
DataStream<Event> stream1 = streamSource1.map(data -> {
String[] fields = data.split(" ");
return new Event(fields[0].trim(), fields[1].trim(), Long.valueOf(fields[2].trim()));
}).assignTimestampsAndWatermarks(WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ofSeconds(2))
.withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
@Override
public long extractTimestamp(Event element, long recordTimestamp) {
return element.timestamp;
}
}));
DataStream<Event> stream2 = streamSource2.map(data -> {
String[] fields = data.split(" ");
return new Event(fields[0].trim(), fields[1].trim(), Long.valueOf(fields[2].trim()));
}).assignTimestampsAndWatermarks(WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ofSeconds(5))
.withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
@Override
public long extractTimestamp(Event element, long recordTimestamp) {
return element.timestamp;
}
}));
//Union合流操作
stream1.union(stream2)
.process(new ProcessFunction<Event, String>() {
@Override
public void processElement(Event value, ProcessFunction<Event, String>.Context ctx, Collector<String> out) throws Exception {
out.collect("水位线:" + ctx.timerService().currentWatermark());
}
}).print();
stream1.print("stream1");
stream2.print("stream2");
env.execute();
}
}
Loading...
马建仓 AI 助手
尝试更多
代码解读
代码找茬
代码优化
1
https://gitee.com/luan_hao/Flink_1.13.git
git@gitee.com:luan_hao/Flink_1.13.git
luan_hao
Flink_1.13
Flink_1.13
master

搜索帮助