本文為英文版的機器翻譯版本，如內容有任何歧義或不一致之處，概以英文版為準。
# 使用 連線至 Amazon ECS `ECSOperator`
本主題說明如何使用 從 Amazon MWAA `ECSOperator`連線至 Amazon Elastic Container Service (Amazon ECS) 容器。在下列步驟中，您將新增必要的許可到環境的執行角色、使用 CloudFormation 範本建立 Amazon ECS Fargate 叢集，最後建立並上傳連線至新叢集的 DAG。
**Topics**
+ [版本](#samples-ecs-operator-version)
+ [先決條件](#samples-ecs-operator-prereqs)
+ [許可](#samples-ecs-operator-permissions)
+ [建立 Amazon ECS 叢集](#create-cfn-template)
+ [範例程式碼](#samples-ecs-operator-code)
## 版本
您可以使用此頁面上的程式碼範例搭配 Python 3.10 中的 **Apache Airflow v2** 和 Python 3.11 中的 **Apache Airflow v**3。 [https://peps.python.org/pep-0619/](https://peps.python.org/pep-0619/) [https://peps.python.org/pep-0664/](https://peps.python.org/pep-0664/)
## 先決條件
若要使用此頁面上的範例程式碼，您需要下列項目：
+ [Amazon MWAA 環境](get-started.md)。
## 許可
+ 您環境的執行角色需要許可，才能在 Amazon ECS 中執行任務。您可以將 [AmazonECS\$1FullAccess](https://console.aws.amazon.com/iam/home#policies/arn:aws:iam::aws:policy/AmazonECS_FullAccess$jsonEditor) 受 AWS管政策連接至您的執行角色，或建立下列政策並將其連接至您的執行角色。
------
#### [ JSON ]
****
```
{
"Version":"2012-10-17",
"Statement": [
{
"Sid": "VisualEditor0",
"Effect": "Allow",
"Action": [
"ecs:RunTask",
"ecs:DescribeTasks"
],
"Resource": "*"
},
{
"Action": "iam:PassRole",
"Effect": "Allow",
"Resource": [
"*"
],
"Condition": {
"StringLike": {
"iam:PassedToService": "ecs-tasks.amazonaws.com"
}
}
}
]
}
```
------
+ 除了新增在 Amazon ECS 中執行任務所需的許可之外，您還必須修改 Amazon MWAA 執行角色中的 CloudWatch Logs 政策陳述式，以允許存取 Amazon ECS 任務日誌群組，如下所示。Amazon ECS 日誌群組是由 CloudFormation 範本在 中建立[建立 Amazon ECS 叢集](#create-cfn-template)。
```
{
"Effect": "Allow",
"Action": [
"logs:CreateLogStream",
"logs:CreateLogGroup",
"logs:PutLogEvents",
"logs:GetLogEvents",
"logs:GetLogRecord",
"logs:GetLogGroupFields",
"logs:GetQueryResults"
],
"Resource": [
"arn:aws:logs:us-east-1:123456789012:log-group:airflow-environment-name-*",
"arn:aws:logs:*:*:log-group:ecs-mwaa-group:*"
]
}
```
如需 Amazon MWAA 執行角色以及如何連接政策的詳細資訊，請參閱 [執行角色](mwaa-create-role.md)。
## 建立 Amazon ECS 叢集
使用下列 CloudFormation 範本，您將建置 Amazon ECS Fargate 叢集以搭配 Amazon MWAA 工作流程使用。如需詳細資訊，請參閱《*Amazon Elastic Container Service 開發人員指南*》中的[建立任務定義](https://docs.aws.amazon.com/AmazonECS/latest/developerguide/create-task-definition)。
1. 使用下列程式碼建立 JSON 檔案，並將其儲存為 `ecs-mwaa-cfn.json`。
```
{
"AWSTemplateFormatVersion": "2010-09-09",
"Description": "This template deploys an ECS Fargate cluster with an Amazon Linux image as a test for MWAA.",
"Parameters": {
"VpcId": {
"Type": "AWS::EC2::VPC::Id",
"Description": "Select a VPC that allows instances access to ECR, as used with MWAA."
},
"SubnetIds": {
"Type": "List",
"Description": "Select at two private subnets in your selected VPC, as used with MWAA."
},
"SecurityGroups": {
"Type": "List",
"Description": "Select at least one security group in your selected VPC, as used with MWAA."
}
},
"Resources": {
"Cluster": {
"Type": "AWS::ECS::Cluster",
"Properties": {
"ClusterName": {
"Fn::Sub": "${AWS::StackName}-cluster"
}
}
},
"LogGroup": {
"Type": "AWS::Logs::LogGroup",
"Properties": {
"LogGroupName": {
"Ref": "AWS::StackName"
},
"RetentionInDays": 30
}
},
"ExecutionRole": {
"Type": "AWS::IAM::Role",
"Properties": {
"AssumeRolePolicyDocument": {
"Statement": [
{
"Effect": "Allow",
"Principal": {
"Service": "ecs-tasks.amazonaws.com"
},
"Action": "sts:AssumeRole"
}
]
},
"ManagedPolicyArns": [
"arn:aws:iam::aws:policy/service-role/AmazonECSTaskExecutionRolePolicy"
]
}
},
"TaskDefinition": {
"Type": "AWS::ECS::TaskDefinition",
"Properties": {
"Family": {
"Fn::Sub": "${AWS::StackName}-task"
},
"Cpu": 2048,
"Memory": 4096,
"NetworkMode": "awsvpc",
"ExecutionRoleArn": {
"Ref": "ExecutionRole"
},
"ContainerDefinitions": [
{
"Name": {
"Fn::Sub": "${AWS::StackName}-container"
},
"Image": "137112412989.dkr.ecr.us-east-1.amazonaws.com/amazonlinux:latest",
"PortMappings": [
{
"Protocol": "tcp",
"ContainerPort": 8080,
"HostPort": 8080
}
],
"LogConfiguration": {
"LogDriver": "awslogs",
"Options": {
"awslogs-region": {
"Ref": "AWS::Region"
},
"awslogs-group": {
"Ref": "LogGroup"
},
"awslogs-stream-prefix": "ecs"
}
}
}
],
"RequiresCompatibilities": [
"FARGATE"
]
}
},
"Service": {
"Type": "AWS::ECS::Service",
"Properties": {
"ServiceName": {
"Fn::Sub": "${AWS::StackName}-service"
},
"Cluster": {
"Ref": "Cluster"
},
"TaskDefinition": {
"Ref": "TaskDefinition"
},
"DesiredCount": 1,
"LaunchType": "FARGATE",
"PlatformVersion": "1.3.0",
"NetworkConfiguration": {
"AwsvpcConfiguration": {
"AssignPublicIp": "ENABLED",
"Subnets": {
"Ref": "SubnetIds"
},
"SecurityGroups": {
"Ref": "SecurityGroups"
}
}
}
}
}
}
}
```
1. 在您的命令提示字元中，使用下列 AWS CLI 命令來建立新的堆疊。您必須將 `SecurityGroups`和 值取代`SubnetIds`為 Amazon MWAA 環境安全群組和子網路的值。
```
aws cloudformation create-stack \
--stack-name my-ecs-stack --template-body file://ecs-mwaa-cfn.json \
--parameters ParameterKey=SecurityGroups,ParameterValue=your-mwaa-security-group \
ParameterKey=SubnetIds,ParameterValue=your-mwaa-subnet-1\\,your-mwaa-subnet-1 \
--capabilities CAPABILITY_IAM
```
或者，您可以使用下列 shell 指令碼。指令碼會使用 `[get-environment](https://awscli.amazonaws.com/v2/documentation/api/latest/reference/mwaa/get-environment.html)` AWS CLI 命令擷取您環境安全群組和子網路的必要值，然後相應地建立堆疊。若要執行指令碼，請執行下列動作。
1. 複製指令碼，並將指令碼儲存為與 CloudFormation 範本`ecs-stack-helper.sh`相同的目錄。
```
#!/bin/bash
joinByString() {
local separator="$1"
shift
local first="$1"
shift
printf "%s" "$first" "${@/#/$separator}"
}
response=$(aws mwaa get-environment --name $1)
securityGroupId=$(echo "$response" | jq -r '.Environment.NetworkConfiguration.SecurityGroupIds[]')
subnetIds=$(joinByString '\,' $(echo "$response" | jq -r '.Environment.NetworkConfiguration.SubnetIds[]'))
aws cloudformation create-stack --stack-name $2 --template-body file://ecs-cfn.json \
--parameters ParameterKey=SecurityGroups,ParameterValue=$securityGroupId \
ParameterKey=SubnetIds,ParameterValue=$subnetIds \
--capabilities CAPABILITY_IAM
```
1. 使用以下命令執行指令碼。將 `environment-name`和 取代`stack-name`為您的資訊。
```
chmod +x ecs-stack-helper.sh
./ecs-stack-helper.bash environment-name stack-name
```
如果成功，您會參考以下顯示新 CloudFormation 堆疊 ID 的輸出。
```
{
"StackId": "arn:aws:cloudformation:us-west-2:123456789012:stack/my-ecs-stack/123456e7-8ab9-01cd-b2fb-36cce63786c9"
}
```
CloudFormation 堆疊完成並 AWS 佈建 Amazon ECS 資源後，您就可以建立和上傳 DAG。
## 範例程式碼
1. 開啟命令提示，然後導覽至存放 DAG 程式碼的目錄。例如：
```
cd dags
```
1. 複製下列程式碼範例的內容，並在本機儲存為 `mwaa-ecs-operator.py`，然後將新的 DAG 上傳至 Amazon S3。
```
from http import client
from airflow import DAG
from airflow.providers.amazon.aws.operators.ecs import ECSOperator
from airflow.utils.dates import days_ago
import boto3
CLUSTER_NAME="mwaa-ecs-test-cluster" #Replace value for CLUSTER_NAME with your information.
CONTAINER_NAME="mwaa-ecs-test-container" #Replace value for CONTAINER_NAME with your information.
LAUNCH_TYPE="FARGATE"
with DAG(
dag_id = "ecs_fargate_dag",
schedule_interval=None,
catchup=False,
start_date=days_ago(1)
) as dag:
client=boto3.client('ecs')
services=client.list_services(cluster=CLUSTER_NAME,launchType=LAUNCH_TYPE)
service=client.describe_services(cluster=CLUSTER_NAME,services=services['serviceArns'])
ecs_operator_task = ECSOperator(
task_id = "ecs_operator_task",
dag=dag,
cluster=CLUSTER_NAME,
task_definition=service['services'][0]['taskDefinition'],
launch_type=LAUNCH_TYPE,
overrides={
"containerOverrides":[
{
"name":CONTAINER_NAME,
"command":["ls", "-l", "/"],
},
],
},
network_configuration=service['services'][0]['networkConfiguration'],
awslogs_group="mwaa-ecs-zero",
awslogs_stream_prefix=f"ecs/{CONTAINER_NAME}",
)
```
**注意**
在範例 DAG 中，對於 `awslogs_group`，您可能需要使用 Amazon ECS 任務日誌群組的名稱來修改日誌群組。此範例假設名為 的日誌群組`mwaa-ecs-zero`。對於 `awslogs_stream_prefix`，請使用 Amazon ECS 任務日誌串流字首。此範例假設日誌串流字首 `ecs`。
1. 執行下列 AWS CLI 命令，將 DAG 複製到您環境的儲存貯體，然後使用 Apache Airflow UI 觸發 DAG。
```
aws s3 cp your-dag.py s3://your-environment-bucket/dags/
```
1. 如果成功，您會在 `ecs_fargate_dag` DAG 中的 任務日誌`ecs_operator_task`中取得類似下列的輸出：
```
[2022-01-01, 12:00:00 UTC] {{ecs.py:300}} INFO - Running ECS Task -
Task definition: arn:aws:ecs:us-west-2:123456789012:task-definition/mwaa-ecs-test-task:1 - on cluster mwaa-ecs-test-cluster
[2022-01-01, 12:00:00 UTC] {{ecs-operator-test.py:302}} INFO - ECSOperator overrides:
{'containerOverrides': [{'name': 'mwaa-ecs-test-container', 'command': ['ls', '-l', '/']}]}
.
.
.
[2022-01-01, 12:00:00 UTC] {{ecs.py:379}} INFO - ECS task ID is: e012340b5e1b43c6a757cf012c635935
[2022-01-01, 12:00:00 UTC] {{ecs.py:313}} INFO - Starting ECS Task Log Fetcher
[2022-01-01, 12:00:00 UTC] {{ecs.py:119}} INFO - [2022-07-19, 17:54:03 UTC] total 52
[2022-01-01, 12:00:00 UTC] {{ecs.py:119}} INFO - [2022-07-19, 17:54:03 UTC] lrwxrwxrwx 1 root root 7 Jun 13 18:51 bin -> usr/bin
[2022-01-01, 12:00:00 UTC] {{ecs.py:119}} INFO - [2022-07-19, 17:54:03 UTC] dr-xr-xr-x 2 root root 4096 Apr 9 2019 boot
[2022-01-01, 12:00:00 UTC] {{ecs.py:119}} INFO - [2022-07-19, 17:54:03 UTC] drwxr-xr-x 5 root root 340 Jul 19 17:54 dev
[2022-01-01, 12:00:00 UTC] {{ecs.py:119}} INFO - [2022-07-19, 17:54:03 UTC] drwxr-xr-x 1 root root 4096 Jul 19 17:54 etc
[2022-01-01, 12:00:00 UTC] {{ecs.py:119}} INFO - [2022-07-19, 17:54:03 UTC] drwxr-xr-x 2 root root 4096 Apr 9 2019 home
[2022-01-01, 12:00:00 UTC] {{ecs.py:119}} INFO - [2022-07-19, 17:54:03 UTC] lrwxrwxrwx 1 root root 7 Jun 13 18:51 lib -> usr/lib
[2022-01-01, 12:00:00 UTC] {{ecs.py:119}} INFO - [2022-07-19, 17:54:03 UTC] lrwxrwxrwx 1 root root 9 Jun 13 18:51 lib64 -> usr/lib64
[2022-01-01, 12:00:00 UTC] {{ecs.py:119}} INFO - [2022-07-19, 17:54:03 UTC] drwxr-xr-x 2 root root 4096 Jun 13 18:51 local
[2022-01-01, 12:00:00 UTC] {{ecs.py:119}} INFO - [2022-07-19, 17:54:03 UTC] drwxr-xr-x 2 root root 4096 Apr 9 2019 media
[2022-01-01, 12:00:00 UTC] {{ecs.py:119}} INFO - [2022-07-19, 17:54:03 UTC] drwxr-xr-x 2 root root 4096 Apr 9 2019 mnt
[2022-01-01, 12:00:00 UTC] {{ecs.py:119}} INFO - [2022-07-19, 17:54:03 UTC] drwxr-xr-x 2 root root 4096 Apr 9 2019 opt
[2022-01-01, 12:00:00 UTC] {{ecs.py:119}} INFO - [2022-07-19, 17:54:03 UTC] dr-xr-xr-x 103 root root 0 Jul 19 17:54 proc
[2022-01-01, 12:00:00 UTC] {{ecs.py:119}} INFO - [2022-07-19, 17:54:03 UTC] dr-xr-x-\-\- 2 root root 4096 Apr 9 2019 root
[2022-01-01, 12:00:00 UTC] {{ecs.py:119}} INFO - [2022-07-19, 17:54:03 UTC] drwxr-xr-x 2 root root 4096 Jun 13 18:52 run
[2022-01-01, 12:00:00 UTC] {{ecs.py:119}} INFO - [2022-07-19, 17:54:03 UTC] lrwxrwxrwx 1 root root 8 Jun 13 18:51 sbin -> usr/sbin
[2022-01-01, 12:00:00 UTC] {{ecs.py:119}} INFO - [2022-07-19, 17:54:03 UTC] drwxr-xr-x 2 root root 4096 Apr 9 2019 srv
[2022-01-01, 12:00:00 UTC] {{ecs.py:119}} INFO - [2022-07-19, 17:54:03 UTC] dr-xr-xr-x 13 root root 0 Jul 19 17:54 sys
[2022-01-01, 12:00:00 UTC] {{ecs.py:119}} INFO - [2022-07-19, 17:54:03 UTC] drwxrwxrwt 2 root root 4096 Jun 13 18:51 tmp
[2022-01-01, 12:00:00 UTC] {{ecs.py:119}} INFO - [2022-07-19, 17:54:03 UTC] drwxr-xr-x 13 root root 4096 Jun 13 18:51 usr
[2022-01-01, 12:00:00 UTC] {{ecs.py:119}} INFO - [2022-07-19, 17:54:03 UTC] drwxr-xr-x 18 root root 4096 Jun 13 18:52 var
.
.
.
[2022-01-01, 12:00:00 UTC] {{ecs.py:328}} INFO - ECS Task has been successfully executed
```