Apache Flink将S3用于后端状态和检查点 背景步骤

问题描述

背景

• 我打算使用S3通过FsStateBackend存储Flink的检查点。但是不知何故我遇到了以下错误。

错误

org.apache.flink.core.fs.UnsupportedFileSystemSchemeException: Could not find a file system implementation for scheme 's3'. The scheme is not directly supported by Flink and no Hadoop file system to support this scheme could be loaded.

Flink版本：：我正在使用Flink 1.10.0版本。

解决方法

我已经找到了解决上述问题的方法，所以在这里，我将按所需步骤列出它。

步骤

1. 我们需要在下面列出的flink-conf.yaml文件中添加一些配置。

state.backend: filesystem
state.checkpoints.dir: s3://s3-bucket/checkpoints/ #"s3://<your-bucket>/<endpoint>"
state.backend.fs.checkpointdir: s3://s3-bucket/checkpoints/ #"s3://<your-bucket>/<endpoint>"


s3.access-key: XXXXXXXXXXXXXXXXXXX #your-access-key
s3.secret-key: xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx #your-secret-key

s3.endpoint: http://127.0.0.1:9000 #your-endpoint-hostname (I have used Minio) 

1. 完成第一步后，我们需要将（flink-s3-fs-hadoop-1.10.0.jar和flink-s3-fs-presto-1.10.0.jar）个JAR文件从opt目录复制到Flink的plugins目录。

   • 例如：-> 1。将/flink-1.10.0/opt/flink-s3-fs-hadoop-1.10.0.jar复制到/flink-1.10.0/plugins/s3-fs-hadoop/flink-s3-fs-hadoop-1.10.0.jar //推荐用于StreamingFileSink
     2。。将/flink-1.10.0/opt/flink-s3-fs-presto-1.10.0.jar复制到/flink-1.10.0/plugins/s3-fs-presto/flink-s3-fs-presto-1.10.0.jar //推荐用于检查点

2. 在检查点代码中添加它

env.setStateBackend(new FsStateBackend("s3://s3-bucket/checkpoints/"))

1. 完成上述所有步骤后，如果Flink已在运行，请重新启动。

注意：

• 如果您同时在Flink中同时使用（flink-s3-fs-hadoop和flink-s3-fs-presto），请专门为s3p://使用flink-s3-fs-presto，为s3a://使用flink-s3-fs-hadoop而不是s3://。
• 有关更多详细信息，请单击here。