如何使用 Apache Airflow 配置为 Celery 设置 SQS 预定义队列？答案

【问题标题】：How to set SQS predefined_queues for Celery with Apache Airflow configuration?如何使用 Apache Airflow 配置为 Celery 设置 SQS 预定义队列？
【发布时间】：2023-03-04 02:42:02
【问题描述】：

我正在使用 celery 和 Amazon SQS 配置 Apache Airflow。我知道 Celery 允许使用 broker_transport_options (https://docs.celeryproject.org/en/stable/getting-started/brokers/sqs.html)，并且 Airflow 在其配置中包含一个名为 celery_broker_transport_options 的部分。

我知道我可以在 Airflow 配置中传递简单的字符串。例如，在 celery_broker_transport 部分，我可以通过：

region = us-west-1

这相当于对芹菜说：

broker_transport_options = {'region': 'us-west-1'}

我正在尝试在 Airflow 中传递预定义队列选项，它在 Celery 中如下所示：

broker_transport_options = {
    'predefined_queues': {
        'my-q': {
            'url': 'https://ap-southeast-2.queue.amazonaws.com/123456/my-q',
            'access_key_id': 'xxx',
            'secret_access_key': 'xxx',
        }
    }
}

我不确定如何将此信息传递给 Airflow。我尝试了以下方法，但我收到一条错误消息，指出“str”对象没有属性“items”：

predefined_queues = 'my-q': { 'url': 'https://sqs.us-east-1.amazonaws.com/1234567890/my-q', }

【问题讨论】：

标签： celery airflow amazon-sqs

【解决方案1】：

您不能直接通过气流配置文件设置此值。相反，您需要使用celery_config_options configuration value 指向在python 代码中设置predefined_queues 的模块。

你可能想要这样的东西：

from airflow.config_templates.default_celery import DEFAULT_CELERY_CONFIG

CELERY_CONFIG = {
    **DEFAULT_CELERY_CONFIG,
    "broker_transport_options": {
        **DEFAULT_CELERY_CONFIG["broker_transport_options"],
        "predefined_queues": {
            "my-q": { "url": "https://sqs.us-east-1.amazonaws.com/1234567890/my-q" },
        },
    },
}

如果你把它放在一个名为 celery_config.py 的文件中，那么你应该能够在你的配置文件中设置 celery_config_options = celery_config.CELERY_CONFIG 并正确配置 celery。

【讨论】：