在flink中保存checkpoint到hdfs中怎么做给个示例

在Flink中将checkpoint保存到HDFS中，需要在Flink作业配置文件中进行配置。具体步骤如下：

1. 在作业配置文件中添加如下配置：

# 设置checkpoint保存到HDFS中
state.backend: filesystem
state.checkpoints.dir: hdfs://<hdfs-namenode>:<port>/<path-to-checkpoint-dir>

其中，<hdfs-namenode>是HDFS的NameNode的地址，<port>是HDFS的端口号，<path-to-checkpoint-dir>是HDFS上存储checkpoint的目录路径。

2. 在代码中启用checkpoint：

// 开启checkpoint，并设置checkpoint的时间间隔为10秒
env.enableCheckpointing(10000);

// 设置checkpoint的保存模式为EXACTLY_ONCE
env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);

// 设置checkpoint的超时时间为1分钟
env.getCheckpointConfig().setCheckpointTimeout(60000);

// 设置同时进行的checkpoint的最大数量为1
env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);

// 设置checkpoint的保存路径
env.setStateBackend(new FsStateBackend("hdfs://<hdfs-namenode>:<port>/<path-to-checkpoint-dir>"));

在代码中，我们首先启用了checkpoint，并设置了checkpoint的时间间隔为10秒，保存模式为EXACTLY_ONCE，超时时间为1分钟，同时进行的checkpoint的最大数量为1。然后，我们设置了checkpoint的保存路径为HDFS上的路径。

完整示例代码如下：

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.contrib.streaming.state.RocksDBStateBackend;
import org.apache.flink.runtime.state.StateBackend;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.PrintSinkFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import java.util.Properties;

public class KafkaConsumerExample {

    public static void main(String[] args) throws Exception {

        // 解析命令行参数
        final ParameterTool params = ParameterTool.fromArgs(args);

        // 设置执行环境
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 设置checkpoint保存到HDFS中
        env.setStateBackend(new FsStateBackend("hdfs://<hdfs-namenode>:<port>/<path-to-checkpoint-dir>"));
        env.enableCheckpointing(10000);
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointTimeout(60000);
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);

        // 设置Kafka参数
        Properties kafkaProps = new Properties();
        kafkaProps.setProperty("bootstrap.servers", params.get("bootstrap.servers", "localhost:9092"));
        kafkaProps.setProperty("group.id", params.get("group.id", "test-group"));

        // 创建Kafka消费者
        FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<>(
                params.get("input-topic", "test-topic"),
                new SimpleStringSchema(),
                kafkaProps
        );

        // 添加Kafka消费者到执行环境中
        DataStream<String> stream = env.addSource(consumer);

        // 对每条消息进行转换，并输出到控制台
        stream.map((MapFunction<String, String>) value -> "Received: " + value)
                .addSink(new PrintSinkFunction<>());

        // 执行作业
        env.execute("Kafka Consumer Example");
    }
}