Azure Flink 检查点到 Azure 存储：找不到帐户的凭据

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 门户上创建存储帐户。我在上面的代码中使用了访问密钥。

当我将应用程序部署到 Kubernetes 时，JobManager 会运行并在 Azure 存储容器中创建检查点文件夹，但是，块 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 协议。我还尝试在 Azure 存储上轮换访问密钥，结果都是相同的错误。

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。

我在启用 Azure 存储插件的情况下将我的 Flink 集群部署到 Kubernetes，如下所示：

./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