如何将 Minio 连接添加到 Airflow 连接？

Question

我正在尝试将 MinIO 的运行实例添加到 Airflow 连接，我认为它应该像 GUI 中的设置一样简单（不用介意公开的凭据，这是环境的阻塞，之后将被更改）：

Airflow 和 minio 都在 docker 容器中运行，它们都使用相同的 docker 网络。按测试按钮会导致以下错误：

测试连接时发生“ClientError”错误：调用 GetCallerIdentity 操作时发生错误 (InvalidClientTokenId)：请求中包含的安全令牌无效。

我很好奇我错过了什么。我们的想法是建立此连接，然后使用存储桶进行数据感知调度（=我想在有人将文件上传到存储桶时立即触发 DAG）

Answer 1

我也面临端点 URL 拒绝连接的问题。我所做的是实际上在 docker 容器中运行，所以我们应该给出 docker 主机 url

{ "aws_access_key_id":"your_minio_access_key", "aws_secret_access_key": "your_minio_secret_key", “主机”：“http://host.docker.internal:9000” }

Answer 2

我在 Airflow 2.5.0 中也遇到了这个错误。 我找到了使用已经内置的 boto3 库的解决方法。

首先我创建了带参数的连接：

连接 ID：任何标签（在我的例子中为 Minio）

连接类型：通用

主机：minio服务器ip和端口

登录：Minio 访问密钥

密码：Minio秘钥

这是我的代码：

import boto3
from airflow.hooks.base import BaseHook

conn = BaseHook.get_connection('Minio')

s3 = boto3.resource('s3',
                     endpoint_url=conn.host,
                     aws_access_key_id=conn.login,
                     aws_secret_access_key=conn.password
)
s3client = s3.meta.client 

#and then you can use boto3 methods for manipulating buckets and files
#for example:

bucket = s3.Bucket('test-bucket')
# Iterates through all the objects, doing the pagination for you. Each obj
# is an ObjectSummary, so it doesn't contain the body. You'll need to call
# get to get the whole body.
for obj in bucket.objects.all():
    key = obj.key

Answer 3

我也面临同样的问题，在将详细信息放入创建连接的网页中后无法测试连接。似乎连接在 DAG 运行期间有效，但在 Web UI 中的测试连接期间失败，在气流亚马逊提供商的Wiki 页面

中也发现了相同的内容

重大变更警告

在此版本的提供商 Amazon S3 连接 (conn_type="s3") 中 由于它始终是 AWS 连接的别名而被删除 conn_type="aws" 实际上，唯一的影响是您将无法 在 Web UI / API 中测试连接。为了恢复能力 要测试连接，您需要更改 Amazon S3 的连接类型 (conn_type="s3") 手动连接至 Amazon Web Services (conn_type="aws")。

Answer 4

已经很晚了，但如果有人遇到连接问题：

即使测试连接失败，您也可以运行 DAG。如果您使用的是 Mac 甚至 Windows，并且您的服务位于默认桥接网络上，那么您应该使用 docker 内部地址主机 IP 解析或获取容器 IP 地址（不建议，因为它是动态的）。

确保 AWS 访问密钥 ID 和 AWS 秘密访问密钥字段均为空，并且仅提供额外的字段。

这是额外字段详细信息的示例：

{ "aws_access_key_id": "xxxxxxx", "aws_secret_access_key": "xxxxxx", "endpoint_url": "http://host.docker.internal:9000" }

注意，endpoint_url 可以替换为 host。

此外，我假设您的airflow和minio正在默认桥接网络上运行，-p 9000:9000并且位于同一个docker-compose.yaml文件中或像这样一起指定

docker-compose -f apache_file.yaml -f minio_file.yaml up

。

如果它们位于用户定义的桥接网络上，则无法使用 host.docker.internal，您必须使用 minio 容器名称/应用程序名称/定义静态地址。您可以发布也可以不发布端口，只需公开它即可。

如果它们位于不同的用户定义的桥接网络上，那么您正在做的事情比我在这里可以解释的更复杂

如果 minio 不在 docker 中，那么你需要获取主机 IP 地址和端口（127.0.0.1:9000）。

如果您在云中运行设置（例如 EC2），那么您需要使用用户定义的桥接网络。

Answer 5

使用 UI 中的测试连接按钮时，它会调用 AWS Security Token Service API GetCallerIdentity。根据文档，并非所有 S3 兼容服务都支持此功能，请参阅 Apache Airflow 文档。

如果您使用 Amazon Provider 与 AWS API 兼容的通信 服务（MinIO、LocalStack 等）测试连接失败不会 意味着您的连接凭据错误。许多兼容 服务仅提供有限数量的 AWS API 服务，并且大多数 其中没有实现 AWS STS GetCallerIdentity 方法。