Amazon 弹性容器服务 (ECS)¶
Amazon Elastic Container Service (Amazon ECS) 是一项全托管的容器编排服务，能够帮助您轻松部署、管理和扩展容器化应用程序。
Airflow 提供运算符用于在 ECS 集群上运行任务定义（Task Definition）。
前置任务¶
要使用这些操作符，您需要完成以下几项工作
通过 pip 安装 API 库。
pip install 'apache-airflow[amazon]'详细信息可在 Airflow® 安装 中获取。
设置连接.
通用参数¶
- aws_conn_id
引用到 Amazon Web Services 连接 ID。如果此参数设置为
None，则使用默认的 boto3 行为且不进行连接查找。否则使用存储在连接中的凭证。默认值：aws_default- region_name
AWS 区域名称。如果此参数设置为
None或省略，则使用 region_name（来自 AWS 连接额外参数）。否则使用指定的值而不是连接中的值。默认值：None- verify
是否验证 SSL 证书。
False- 不验证 SSL 证书。path/to/cert/bundle.pem - 要使用的 CA 证书捆绑文件名。如果您想使用与 botocore 不同的 CA 证书捆绑，可指定此参数。
如果此参数设置为
None或省略，则使用 verify（来自 AWS 连接额外参数）。否则使用指定的值而不是连接中的值。默认值：None- botocore_config
提供的字典将用于构建 botocore.config.Config。此配置可用于配置避免限流异常、超时等。
示例，关于参数的更多细节请查看 botocore.config.Config¶{ "signature_version": "unsigned", "s3": { "us_east_1_regional_endpoint": True, }, "retries": { "mode": "standard", "max_attempts": 10, }, "connect_timeout": 300, "read_timeout": 300, "tcp_keepalive": True, }
如果此参数设置为
None或省略，则使用 config_kwargs（来自 AWS 连接额外参数）。否则使用指定的值而不是连接中的值。默认值：None注意
指定一个空字典
{}将覆盖 botocore.config.Config 的连接配置。
操作符¶
创建一个 AWS ECS 集群¶
要创建 Amazon ECS 集群，您可以使用 EcsCreateClusterOperator。
传递给 Create Cluster API 的所有可选参数应放在 ‘create_cluster_kwargs’ 字典中。
create_cluster = EcsCreateClusterOperator(
task_id="create_cluster",
cluster_name=new_cluster_name,
)
删除一个 AWS ECS 集群¶
要删除 Amazon ECS 集群，您可以使用 EcsDeleteClusterOperator。
delete_cluster = EcsDeleteClusterOperator(
task_id="delete_cluster",
cluster_name=new_cluster_name,
)
注册任务定义¶
要注册任务定义，您可以使用 EcsRegisterTaskDefinitionOperator。
传递给 Register Task Definition API 的所有可选参数应放在 ‘register_task_kwargs’ 字典中。
register_task = EcsRegisterTaskDefinitionOperator(
task_id="register_task",
family=family_name,
container_definitions=[
{
"name": container_name,
"image": "ubuntu",
"workingDirectory": "/usr/bin",
"entryPoint": ["sh", "-c"],
"command": ["ls"],
"logConfiguration": {
"logDriver": "awslogs",
"options": {
"awslogs-group": log_group_name,
"awslogs-region": aws_region,
"awslogs-create-group": "true",
"awslogs-stream-prefix": "ecs",
},
},
},
],
register_task_kwargs={
"cpu": "256",
"memory": "512",
"networkMode": "awsvpc",
},
)
Deregister a Task Definition¶
要注销任务定义，您可以使用 EcsDeregisterTaskDefinitionOperator。
deregister_task = EcsDeregisterTaskDefinitionOperator(
task_id="deregister_task",
task_definition=register_task.output,
)
运行任务定义¶
要在 Amazon ECS 集群中运行已定义的任务定义，您可以使用 EcsRunTaskOperator。
在使用此运算符之前，您需要先创建 ECS 集群并注册任务定义。任务定义中包含您想要运行的容器化应用程序的详细信息。
此运算符支持在以下三种类型的 ECS 集群中运行容器：Serverless（FARGATE）、EC2 或外部资源（EXTERNAL）。您需要配置的参数取决于您想使用的 launch_type。
launch_type="EC2|FARGATE|EXTERNAL"
如果在 ECS 集群中使用 AWS Fargate 作为计算资源，请将参数
launch_type设置为 FARGATE。使用 FARGATE 启动类型时，需要提供network_configuration参数。如果在 ECS 集群中使用 EC2 作为计算资源，请将参数设置为 EC2。
如果您在 ECS 集群中集成了外部资源（例如使用 ECS Anywhere），并希望在这些外部资源上运行容器，请将参数设置为 EXTERNAL。
run_task = EcsRunTaskOperator(
task_id="run_task",
reattach=True,
cluster=existing_cluster_name,
task_definition=register_task.output,
overrides={
"containerOverrides": [
{
"name": container_name,
"command": ["echo hello world"],
},
],
},
network_configuration={"awsvpcConfiguration": {"subnets": existing_cluster_subnets}},
awslogs_group=log_group_name,
awslogs_region=aws_region,
awslogs_stream_prefix=f"ecs/{container_name}",
)
hello_world = EcsRunTaskOperator(
task_id="hello_world",
cluster=cluster_name,
task_definition=task_definition_name,
launch_type="FARGATE",
overrides={
"containerOverrides": [
{
"name": container_name,
"command": ["echo", "hello", "world"],
},
],
},
network_configuration={
"awsvpcConfiguration": {
"subnets": test_context[SUBNETS_KEY],
"securityGroups": test_context[SECURITY_GROUPS_KEY],
"assignPublicIp": "ENABLED",
},
},
)
将日志流式传输至 AWS CloudWatch¶
要将日志流式传输至 AWS CloudWatch，您需要定义以下参数。使用上面的示例时，只需在原有参数基础上添加这些额外参数即可启用 CloudWatch 日志。请确保您拥有相应的权限（见下节）。
awslogs_group=log_group_name,
awslogs_region=aws_region,
awslogs_stream_prefix=f"ecs/{container_name}",
IAM 权限¶
运行 EcsRunTaskOperator 执行任务时，您必须具备以下 IAM 权限。
{
"Effect": "Allow",
"Action": [
"ecs:RunTask",
"ecs:DescribeTasks",
]
"Resource": [ "arn:aws:ecs:{aws region}:{aws account number}:task_definition/{task definition family}" ]
},
{
"Effect": "Allow",
"Action": [
"iam:PassRole"
]
"Resource": [ "arn:aws:iam::{aws account number}:role/{task execution role name}" ]
},
{
"Effect": "Allow",
"Action": [
"ecs:DescribeTasks",
],
"Resource": [ "arn:aws:ecs:{aws region}:{aws account number}:task/{ecs cluster name}/*" ]
}
如果使用 “reattach=True”（默认值为 False），还需要添加额外的权限。请在 IAM 策略中加入以下额外的 Action。
"ecs:DescribeTaskDefinition",
"ecs:ListTasks"
CloudWatch 权限
如果计划将 Apache Airflow 日志流式传输到 AWS CloudWatch，请确保已配置相应的权限集合。
iam.PolicyStatement(
actions=[
"logs:CreateLogStream",
"logs:CreateLogGroup",
"logs:PutLogEvents",
"logs:GetLogEvents",
"logs:GetLogRecord",
"logs:GetLogGroupFields",
"logs:GetQueryResults"
],
effect=iam.Effect.ALLOW,
resources=[
"arn:aws:logs:{aws region}:{aws account number}:log-group:{aws-log-group-name}:log-stream:{aws-log-stream-name}/\*"
]
)
传感器¶
AWS ECS 集群状态传感器¶
要轮询集群状态直至达到终止状态，可使用 EcsClusterStateSensor。
默认成功状态为 EcsClusterStates.ACTIVE，未设定失败状态，两者均可通过参数覆盖。如果在达到目标状态之前进入了提供的失败状态，则会抛出带有失败原因的 AirflowException。
await_cluster = EcsClusterStateSensor(
task_id="await_cluster",
cluster_name=new_cluster_name,
)
AWS ECS 任务定义状态传感器¶
要轮询任务定义状态直至达到终止状态，可使用 EcsTaskDefinitionStateSensor。
有效状态为 EcsTaskDefinitionStates.ACTIVE 或 EcsTaskDefinitionStates.INACTIVE。默认成功状态为 EcsTaskDefinitionStates.ACTIVE，可通过参数更改。若在达到目标状态前进入失败状态，则会抛出带有失败原因的 AirflowException。
await_task_definition = EcsTaskDefinitionStateSensor(
task_id="await_task_definition",
task_definition=register_task.output,
)
AWS ECS 任务状态传感器¶
要轮询任务状态直至达到终止状态，可使用 EcsTaskStateSensor。
默认成功状态为 EcsTaskStates.RUNNING，未设定失败状态，两者均可通过参数覆盖。如果在达到目标状态前进入提供的失败状态，则会抛出带有失败原因的 AirflowException。
# By default, EcsTaskStateSensor waits until the task has started, but the
# demo task runs so fast that the sensor misses it. This sensor instead
# demonstrates how to wait until the ECS Task has completed by providing
# the target_state and failure_states parameters.
await_task_finish = EcsTaskStateSensor(
task_id="await_task_finish",
cluster=cluster_name,
task=hello_world.output["ecs_task_arn"],
target_state=EcsTaskStates.STOPPED,
failure_states={EcsTaskStates.NONE},
)