Flink 停机恢复（savepoint）入门示例

最新推荐文章于 2025-03-07 02:51:15 发布

Mathieu66

最新推荐文章于 2025-03-07 02:51:15 发布

阅读量1.6k

点赞数 1

分类专栏： Apache Flink 文章标签： flink 高可用 savepoint

本文链接：https://blog.csdn.net/Mathieu66/article/details/99756660

版权

Apache Flink 专栏收录该内容

19 篇文章

订阅专栏

本文介绍如何使用Apache Flink实现状态化的单词计数，包括从socket接收数据、按单词分割并计数，以及利用Savepoint进行任务恢复，确保计算从上次停止处继续。

摘要生成于 C知道，由 DeepSeek-R1 满血版支持，前往体验 >

目标：使用flink从socket接收字符串，根据空格分割成单词，统计单词出现的次数。使用savepoint取消任务，恢复，使计算从上一次停止的地方继续。

主程序：

package com.maxiu.state;

import com.maxiu.function.StatefulWordCountFunction;
import com.maxiu.model.WordWithCount;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

/**
 * @author xianghu.wang
 * @time 2019/8/19
 * @description
 */
public class StatefulWordCount {
    public static void main(String[] args) throws Exception {
        // 1.获取执行环境
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 设置检查点时间间隔为1s,取消任务时不删除检查点
        env.enableCheckpointing(1000);
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

        // 2.读取数据：连接到socket获取文本数据
        DataStream<String> text = env.socketTextStream("localhost", 9999, "\n");

        // 3.业务逻辑编码
        DataStream<WordWithCount> windowCounts = text
                .flatMap(new FlatMapFunction<String, WordWithCount>() {
                    @Override
                    public void flatMap(String value, Collector<WordWithCount> out) {
                        for (String word : value.split("\\s")) {
                            out.collect(new WordWithCount(word, 1));
                        }
                    }
                })
                .keyBy(WordWithCount::getWord)
                .timeWindow(Time.seconds(5), Time.seconds(5))
                .process(new StatefulWordCountFunction());


        // 4.输出结果：使用一个线程打印结果
        windowCounts.print().setParallelism(1);

        // 5.开始执行任务
        env.execute("StatefulWordCount");
    }
}

bean类：

package com.ccclubs.model;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.ToString;

/**
 * @author xianghu.wang
 * @time 2019/8/19
 * @description
 */
@Data
@AllArgsConstructor
public class WordWithCount {
    public String word;
    public Integer count;
}

开干：

1、打包：包名 flink-demo-stateful.jar

2、启动 netcat