Flink Fold怎么使用

這篇文章主要講解了“Flink Fold怎么使用”，文中的講解內容簡單清晰，易于學習與理解，下面請大家跟著小編的思路慢慢深入，一起來研究和學習“Flink Fold怎么使用”吧！

Fold算子：將數據流的每一次輸出進行滾動疊加，合并輸出結果

示例環境

java.version: 1.8.x
flink.version: 1.11.1

示例數據源

Flink 系例 之 搭建開發環境與數據

Fold.java

import com.flink.examples.DataSource;
import org.apache.flink.api.common.functions.FoldFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import java.util.List;

/**
 * @Description Fold算子：將數據流的每一次輸出進行滾動疊加，合并輸出結果
 * （與Reduce的區別是，Reduce是拿前一次聚合結果累加后一次的并輸出數據流；Fold是直接將當前數據對象追加到前一次疊加結果上并輸出數據流）
 */
public class Fold {
    /**
     * 遍歷集合，分區打印每一次滾動疊加的結果（示例：按性別分區，按排序，未位追加輸出）
     * @param args
     * @throws Exception
     */
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        List<Tuple3<String,String,Integer>> tuple3List = DataSource.getTuple3ToList();
        //注意：使用Integer進行分區時，會導致分區結果不對，轉換成String類型輸出key即可正確輸出
        KeyedStream<Tuple3<String,String,Integer>, String> keyedStream = env.fromCollection(tuple3List).keyBy(new KeySelector<Tuple3<String,String,Integer>, String>() {
            @Override
            public String getKey(Tuple3<String, String, Integer> tuple3) throws Exception {
                //f1為性別字段,以相同f1值（性別）進行分區
                return String.valueOf(tuple3.f1);
            }
        });
        SingleOutputStreamOperator<String> result = keyedStream.fold("同學：", new FoldFunction<Tuple3<String, String, Integer>, String>() {
            @Override
            public String fold(String s, Tuple3<String, String, Integer> tuple3) throws Exception {
                if (s.startsWith("男") || s.startsWith("女")){
                    return s + tuple3.f0 + "、";
                } else {
                    return (tuple3.f1.equals("man") ? "男" : "女") + s + tuple3.f0 + "、";
                }
            }
        });
        result.print();
        env.execute("flink Fold job");
    }
}

打印結果

2> 男同學：張三、
2> 男同學：張三、王五、
2> 男同學：張三、王五、吳八、
1> 女同學：李四、
1> 女同學：李四、劉六、
1> 女同學：李四、劉六、伍七、

感謝各位的閱讀，以上就是“Flink Fold怎么使用”的內容了，經過本文的學習后，相信大家對Flink Fold怎么使用這一問題有了更深刻的體會，具體使用情況還需要大家實踐驗證。這里是億速云，小編將為大家推送更多相關知識點的文章，歡迎關注！