Azure Flink 检查点到 Azure 存储 : No credentials found for account

Question

我有一个测试 Flink 应用程序，正在尝试在连接到 Azure 存储的 Azure Kubernetes 上运行。在我的 Flink 应用程序中，我配置了以下配置:

Configuration cfg = new Configuration();
cfg.setString("fs.azure.account.key.<storage-account.blob.core.windows.net", "<access-key>");
FileSystem.initialize(cfg, null);

我还启用了检查点，如下所示:

env.enableCheckpointing(10000);
env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.DELETE_ON_CANCELLATION);
env.setStateBackend(new EmbeddedRocksDBStateBackend());
env.getCheckpointConfig().setCheckpointStorage("wasbs://<container>@<storage-account>.blob.core.windows.net/checkpoint/");

已在 Azure 门户上创建存储帐户。我在上面的代码中使用了访问 key 。

当我将应用程序部署到 Kubernetes 时，JobManager 会运行并在 Azure 存储容器中创建检查点文件夹，但是， block blob 数据的大小始终为 0B。应用程序也不断抛出此异常。

我遇到的有趣错误是:

Caused by: org.apache.flink.fs.shaded.hadoop3.org.apache.hadoop.fs.azure.AzureException: No credentials found for account <storage-account>.blob.core.windows.net in the configuration, and its container <container> is not accessible using anonymous credentials. Please check if the container exists first. If it is not publicly available, you have to provide account credentials.

org.apache.flink.fs.azure.shaded.com.microsoft.azure.storage.StorageException: Public access is not permitted on this storage account

让我摸不着头脑的部分(除了跳蚤)是它确实创建了检查点文件夹和文件，并继续创建更多检查点。

此帐户不可公开访问，公司政策限制公开访问。

我还尝试使用 flink-conf.yaml，这是我的示例:

state:backend: rocksdb
state.checkpoints.dir: wasbs://<container>@<storage-account>.blob.core.windows.net/checkpoint/
fs.azure.account.key.**flinkstorage**.blob.core.windows.net: <access-key>
fs.azure.account.key.<storage-account>.blob.core.windows.net: <access-key>

我尝试了上面的两个 account.key 选项。我也尝试过 wasb 协议(protocol)。我还尝试在 Azure 存储上轮换访问 key ，但都导致了相同的错误。

Answer 1

我最终通过将所有检查点配置移至 flink-conf.yaml 来实现此功能。所有对检查点的引用都已从我的代码(即 StreamExecutionEnvironment)中删除。

我的 flink-config.yaml 看起来像这样

execution.checkpointing.interval: 10s
execution.checkpoint.mode: EXACTLY_ONCE
state.backend: rocksdb
state.checkpoints.dir: wasbs://<container>@<storage-account.blob.core.windows.net/checkpoint/

# azure storage access key
fs.azure.account.key.psbombb.blob.core.windows.net: <access-key>

检查点现已写入 Azure 存储，元数据文件的大小不再是 0B。

我将 Flink 集群部署到 Kubernetes，如下所示，并启用了 Azure 存储插件:

./bin/kubernetes-session.sh -Dkubernetes.cluster-id=<cluster-name> -Dkubernetes.namespace=<your-namespace> -Dcontainerized.master.env.ENABLE_BUILT_IN_PLUGINS=flink-azure-fs-hadoop-1.14.0.jar -Dcontainerized.taskmanager.env.ENABLE_BUILT_IN_PLUGINS=flink-azure-fs-hadoop-1.14.0.jar

然后我将作业部署到 Flink 集群，如下所示:

./bin/flink run --target kubernetes-session -Dkubernetes.namespace=<your-namespace> -Dkubernetes.cluster-id=<cluster-name> ~/path/to/project/<your-jar>.jar

WebUI 上的 TaskManager 将不会显示 StdOut 日志。您需要kubectl logs -f <taskmanager-pod-name> -n <your-namespace>查看作业日志。

如果您想查看 Flink WebUI，请记住端口转发 8081: kubectl port-forward svc/<cluster-name> -n <namespace>

例如http://localhost:8081

如果您使用 Minikube 并且希望通过 Flink LoadBalancer 外部 IP 访问集群，您需要运行 minikube tunnel

例如http://<external-ip>:8081