airflow.providers.amazon.aws.hooks.batch_waiters¶
AWS Batch 服务等待器。
另请参阅
类¶
一个用于管理 AWS Batch 服务等待器的工具类。 |
模块内容¶
- class airflow.providers.amazon.aws.hooks.batch_waiters.BatchWaitersHook(*args, waiter_config=None, **kwargs)[source]¶
基类:
airflow.providers.amazon.aws.hooks.batch_client.BatchClientHook
一个用于管理 AWS Batch 服务等待器的工具类。
import random from airflow.providers.amazon.aws.operators.batch_waiters import BatchWaiters # to inspect default waiters waiters = BatchWaiters() config = waiters.default_config # type: Dict waiter_names = waiters.list_waiters() # -> ["JobComplete", "JobExists", "JobRunning"] # The default_config is a useful stepping stone to creating custom waiters, e.g. custom_config = waiters.default_config # this is a deepcopy # modify custom_config['waiters'] as necessary and get a new instance: waiters = BatchWaiters(waiter_config=custom_config) waiters.waiter_config # check the custom configuration (this is a deepcopy) waiters.list_waiters() # names of custom waiters # During the init for BatchWaiters, the waiter_config is used to build a waiter_model; # and note that this only occurs during the class init, to avoid any accidental mutations # of waiter_config leaking into the waiter_model. waiters.waiter_model # -> botocore.waiter.WaiterModel object # The waiter_model is combined with the waiters.client to get a specific waiter # and the details of the config on that waiter can be further modified without any # accidental impact on the generation of new waiters from the defined waiter_model, e.g. waiters.get_waiter("JobExists").config.delay # -> 5 waiter = waiters.get_waiter("JobExists") # -> botocore.waiter.Batch.Waiter.JobExists object waiter.config.delay = 10 waiters.get_waiter("JobExists").config.delay # -> 5 as defined by waiter_model # To use a specific waiter, update the config and call the `wait()` method for jobId, e.g. waiter = waiters.get_waiter("JobExists") # -> botocore.waiter.Batch.Waiter.JobExists object waiter.config.delay = random.uniform(1, 10) # seconds waiter.config.max_attempts = 10 waiter.wait(jobs=[jobId])
另请参阅
- 参数:
waiter_config (dict | None) – AWS Batch 服务的自定义等待器配置
aws_conn_id – AWS 凭证 / 区域名称的连接 ID。如果为 None,将使用 boto3 凭证策略 (https://boto3.amazonaws.com/v1/documentation/api/latest/guide/configuration.html)。
region_name – 在 AWS 客户端中使用的区域名称。覆盖连接中(如果提供)的 AWS 区域。
- property waiter_config: dict[source]¶
此实例的不可变等待器配置;此属性返回一个
deepcopy
。在 BatchWaiters 初始化期间,waiter_config 用于构建 waiter_model,这仅在类初始化期间发生,以避免 waiter_config 的任何意外修改泄露到 waiter_model 中。
- 返回:
AWS Batch 服务的等待器配置
- 返回类型:
- property waiter_model: botocore.waiter.WaiterModel[source]¶
用于在 AWS Batch 服务上生成等待器的已配置等待器模型。
- 返回:
AWS Batch 服务的等待器模型
- 返回类型:
botocore.waiter.WaiterModel
- get_waiter(waiter_name, parameters=None, config_overrides=None, deferrable=False, client=None)[source]¶
使用已配置的
.waiter_model
获取 AWS Batch 服务等待器。.waiter_model
与.client
结合使用以获取特定的等待器,并且可以修改该等待器的属性,而不会对从.waiter_model
生成新等待器产生任何意外影响,例如:waiters.get_waiter("JobExists").config.delay # -> 5 waiter = waiters.get_waiter("JobExists") # a new waiter object waiter.config.delay = 10 waiters.get_waiter("JobExists").config.delay # -> 5 as defined by waiter_model
要使用特定的等待器,请更新配置并调用 jobId 的 wait() 方法,例如:
import random waiter = waiters.get_waiter("JobExists") # a new waiter object waiter.config.delay = random.uniform(1, 10) # seconds waiter.config.max_attempts = 10 waiter.wait(jobs=[jobId])
- 参数:
waiter_name (str) – 等待器的名称。该名称应与等待器模型文件中的键名称(包括大小写)匹配(通常是 CamelCasing);请参阅
.list_waiters
。parameters (dict[str, str] | None) – 未使用,仅用于匹配 base_aws 中的方法签名
config_overrides (dict[str, Any] | None) – 未使用,仅用于匹配 base_aws 中的方法签名
deferrable (bool) – 未使用,仅用于匹配 base_aws 中的方法签名
client – 未使用,仅用于匹配 base_aws 中的方法签名
- 返回:
指定 AWS Batch 服务的等待器对象
- 返回类型:
botocore.waiter.Waiter
- wait_for_job(job_id, delay=None, get_batch_log_fetcher=None)[source]¶
等待 Batch 作业完成。
这假设
.waiter_model
使用.default_config
的某种变体进行配置,以便它可以生成具有以下名称的等待器:“JobExists”、“JobRunning”和“JobComplete”。- 参数:
job_id (str) – Batch 作业 ID
get_batch_log_fetcher (Callable[[str], airflow.providers.amazon.aws.utils.task_log_fetcher.AwsTaskLogFetcher | None] | None) – 一个方法,返回 AwsTaskLogFetcher 类型的 batch_log_fetcher,或者在 CloudWatch 日志流尚未创建时返回 None。
- 抛出:
AirflowException
注意
此方法会给
delay
添加一个小的随机抖动(+/- 2 秒,>= 1 秒)。使用随机间隔有助于在许多并发任务请求作业描述时避免 AWS API 限制。它还将
max_attempts
修改为使用sys.maxsize
,这使得 Airflow 能够管理等待超时。