Map
输入DataStream返回DataStream。
接收一个元素,产生一个元素。下面是转换为双倍值的MapFunction
DataStream<Integer> dataStream = //...
dataStream.map(new MapFunction<Integer, Integer>() {
@Override
public Integer map(Integer value) throws Exception {
return 2 * value;
}
});
FlatMap
输入DataStream返回DataStream。
接收一个元素,产出0个,1个,或者更多的元素。下面是一个字符串拆分为多个字符串的FlatMap
dataStream.flatMap(new FlatMapFunction<String, String>() {
@Override
public void flatMap(String value, Collector<String> out)
throws Exception {
for(String word: value.split(" ")){
out.collect(word);
}
}
});
Filter
输入DataStream返回DataStream。
过滤数据,function返回为true的会被保留,为false的会被排除。
dataStream.filter(new FilterFunction<Integer>() {
@Override
public boolean filter(Integer value) throws Exception {
return value != 0;
}
});
KeyBy
输入DataStream返回KeyedStream。
按照key将数据拆分为不同的集合,具有相同key的数据放到同一个集合,内部使用hashCode来判断是否属于同一个key。
dataStream.keyBy(value -> value.getSomeKey());
dataStream.keyBy(value -> value.f0);
Reduce
输入 KeyedStream 返回 DataStream。
将按照key拆分的集合滚动处理。合并当前元素和最后一次合并的结果,然后返回一个新的值。
keyedStream.reduce(new ReduceFunction<Integer>() {
@Override
public Integer reduce(Integer value1, Integer value2)
throws Exception {
return value1 + value2;
}
});
Window
输入 KeyedStream 返回 WindowedStream。
窗口可以定义在KeyedStreams上,窗口可以将每个key的数据按照某种特征分组,点击链接 windows 查看完整的针对窗口的描述
dataStrea。m
.keyBy(value -> value.f0)
.window(TumblingEventTimeWindows.of(Time.seconds(5)));
WindowAll
输入 DataStream 返回 AllWindowedStream。
窗口可以定义在常规的DataStream上。窗口可以将数据按照某种特征分组,点击链接 windows 查看完整的针对窗口的描述
dataStream
.windowAll(TumblingEventTimeWindows.of(Time.seconds(5)));
Window Apply
输入 WindowedStream或者AllWindowedStream 输出 DataStream 。
应用一个函数reduce到窗口。点击链接 windows 查看完整的针对窗口的描述
windowedStream.apply(new WindowFunction<Tuple2<String,Integer>, Integer, Tuple, Window>() {
public void apply (Tuple tuple,
Window window,
Iterable<Tuple2<String, Integer>> values,
Collector<Integer> out) throws Exception {
int sum = 0;
for (value t: values) {
sum += t.f1;
}
out.collect (new Integer(sum));
}
});
WindowFunction 类型参数:
– 输入值的类型。
– 输出值的类型。
– 密钥的类型。
– 可以应用此窗口函数的Window类型。
// applying an AllWindowFunction on non-keyed window stream
allWindowedStream.apply (new AllWindowFunction<Tuple2<String,Integer>, Integer, Window>() {
public void apply (Window window,
Iterable<Tuple2<String, Integer>> values,
Collector<Integer> out) throws Exception {
int sum = 0;
for (value t: values) {
sum += t.f1;
}
out.collect (new Integer(sum));
}
});
AllWindowFunction类型参数:
– 输入值的类型。
– 输出值的类型。
– 可以应用此窗口函数的Window类型。
WindowReduce
输入 WindowedStream 输出 DataStream 。
应用一个Reduct函数到窗口,并返回合并后的值。
windowedStream.reduce (new ReduceFunction<Tuple2<String,Integer>>() {
public Tuple2<String, Integer> reduce(Tuple2<String, Integer> value1, Tuple2<String, Integer> value2) throws Exception {
return new Tuple2<String,Integer>(value1.f0, value1.f1 + value2.f1);
}
});
Union
合并两个或更多的流,返回新的流包含所有流中的元素。包含重复的。
dataStream.union(otherStream1, otherStream2, ...);
Window Join
基于指定的key和共同窗口join两个数据流,返回一个新的数据流。
dataStream.join(otherStream)
.where(<key selector>).equalTo(<key selector>)
.window(TumblingEventTimeWindows.of(Time.seconds(3)))
.apply (new JoinFunction () {...});
Interval Join
输入 KeyedStream,返回一个数据流。
基于在指定时间间隔内的共同key,Join 两个KeyedStream的流。
// this will join the two streams so that
// key1 == key2 && leftTs - 2 < rightTs < leftTs + 2
keyedStream.intervalJoin(otherKeyedStream)
.between(Time.milliseconds(-2), Time.milliseconds(2)) // lower and upper bound
.upperBoundExclusive(true) // optional
.lowerBoundExclusive(true) // optional
.process(new IntervalJoinFunction() {...});
Window CoGroup
输入两个数据流,返回一个数据流。
将两个流按照指定key和公共窗口合并,某些键可能只包含在两个原始数据集之一中。 在这种情况下,对于不包含具有该特定键的元素的数据集一侧,将使用空输入调用 CoGroup 函数。
CoGroupFunction类型参数:
– 第一个输入数据集的数据类型。
– 第二个输入数据集的数据类型。
– 返回元素的数据类型。
dataStream.coGroup(otherStream)
.where(0).equalTo(1)
.window(TumblingEventTimeWindows.of(Time.seconds(3)))
.apply (new CoGroupFunction () {...});
Connect
输入 DataStream,DataStream ,返回ConnectedStream。
连接两个数据流保持原有类型。连接允许两个流之间共享状态。
DataStream<Integer> someStream = //...
DataStream<String> otherStream = //...
ConnectedStreams<Integer, String> connectedStreams = someStream.connect(otherStream);
CoMap,CoFlatMap
输入 ConnectedStream 输出 DataStream 。
类似于已关联数据流上的map和flatMap。
connectedStreams.map(new CoMapFunction<Integer, String, Boolean>() {
@Override
public Boolean map1(Integer value) {
return true;
}
@Override
public Boolean map2(String value) {
return false;
}
});
connectedStreams.flatMap(new CoFlatMapFunction<Integer, String, String>() {
@Override
public void flatMap1(Integer value, Collector<String> out) {
out.collect(value.toString());
}
@Override
public void flatMap2(String value, Collector<String> out) {
for (String word: value.split(" ")) {
out.collect(word);
}
}
});
Iterate
通过将一个操作符的输出重定向到之前的某个操作符,在流中创建一个“反馈”循环。这对于定义不断更新模型的算法尤其有用。下面的代码从一个流开始,并不断地应用迭代体。大于0的元素被发送回反馈通道,其余的元素被向下转发。
IterativeStream<Long> iteration = initialStream.iterate();
DataStream<Long> iterationBody = iteration.map (/*do something*/);
DataStream<Long> feedback = iterationBody.filter(new FilterFunction<Long>(){
@Override
public boolean filter(Long value) throws Exception {
return value > 0;
}
});
iteration.closeWith(feedback);
DataStream<Long> output = iterationBody.filter(new FilterFunction<Long>(){
@Override
public boolean filter(Long value) throws Exception {
return value <= 0;
}
});
了解 工作生活心情记忆 的更多信息
Subscribe to get the latest posts sent to your email.