Amazon Apache 氣流管理工作流程的快速入門教學

本快速入門教學使用可同時建立 Amazon VPC 基礎設施的 AWS CloudFormation 範本、一個含有dags資料夾的 Amazon S3 儲存貯體，以及適用於 Apache 氣流環境的 Amazon 受管工作流程。

於本教學課程中

本教學會逐步引導您完成三個 AWS Command Line Interface (AWS CLI) 命令，以便將 DAG 上傳至 Amazon S3、在 Apache 氣流中執行 DAG，以及檢視日誌 CloudWatch。最後，我們會逐步引導您為 Apache 氣流開發團隊建立 IAM 政策的步驟。

注意

此頁面上的 AWS CloudFormation 範本會針對中提供的最新版本的 Apache 氣流環境，建立適用於 Apache 氣流環境的 Amazon 受管工作流程 AWS CloudFormation。可用的最新版本是阿帕奇氣流 v2.8.1。

此頁面上的 AWS CloudFormation 範本會建立下列項目：

• VPC 基礎架構。範本使用網際網路上的公用路由。它使用中公共網路存取模式的 Apache 氣流網頁伺服器WebserverAccessMode: PUBLIC_ONLY。

• Amazon S3 桶。該模板創建一個帶有dags文件夾的 Amazon S3 存儲桶。它配置為阻止所有公共訪問，並啟用了存儲桶版本控制，如中的定義為 Amazon MWAA 儲存貯貯存貯貯貯存貯貯貯存貯貯貯。

• Amazon MWAA 環境。該範本會建立與 Amazon S3 儲存貯體上dags資料夾相關聯的 Amazon MWAA 環境、具有 Amazon MWAA 所使用 AWS 服務的許可執行角色，以及使用AWS 擁有金鑰加密的預設角色 (如中所定義)。建立一個 Amazon MWAA 環境

• CloudWatch 記錄檔。範本會針對氣流排程器記錄群組、Airflow Web 伺服器記錄群組、Airflow 工作者記錄群組、Airflow 工作者記錄群組、Airflow DAG 處理記錄群組，以及氣流工作記錄群組 (如中所定義) 啟用 Apache Airflow 登入在 Amazon 中查看氣流日誌 CloudWatch。 CloudWatch

在本教學課程中，您將完成下列工作：

• 上傳並執行 DAG。將 Apache 氣流的教程 DAG 上傳最新的 Amazon MWAA 支持 Apache 氣流版本到 Amazon S3，然後在 Apache 氣流用戶界面中運行，如中所定義。新增或更新 DAG

• 檢視記錄檔。在記錄檔中檢視 Airflow Web 伺服器 CloudWatch 記錄群組，如中所定義在 Amazon 中查看氣流日誌 CloudWatch。

• 建立存取控制原則。根據中的定義，在 IAM 中為您的 Apache 氣流開發團隊建立存取控制政策存取 Amazon MWAA 環境。

必要條件

AWS Command Line Interface (AWS CLI) 是開放原始碼工具，可讓您使用命令列殼層中的命令與 AWS 服務互動。若要完成此頁面上的步驟，您需要下列項目：

• AWS CLI — 安裝版本 2.

• AWS CLI — 快速配置 aws configure.

步驟 1：將 AWS CloudFormation 範本儲存在本機

• 複製以下範本的內容並在本機儲存為mwaa-public-network.yml。您也可以下載模板。

  AWSTemplateFormatVersion: "2010-09-09"
  
  Parameters:
  
    EnvironmentName:
      Description: An environment name that is prefixed to resource names
      Type: String
      Default: MWAAEnvironment
  
    VpcCIDR:
      Description: The IP range (CIDR notation) for this VPC
      Type: String
      Default: 10.192.0.0/16
  
    PublicSubnet1CIDR:
      Description: The IP range (CIDR notation) for the public subnet in the first Availability Zone
      Type: String
      Default: 10.192.10.0/24
  
    PublicSubnet2CIDR:
      Description: The IP range (CIDR notation) for the public subnet in the second Availability Zone
      Type: String
      Default: 10.192.11.0/24
  
    PrivateSubnet1CIDR:
      Description: The IP range (CIDR notation) for the private subnet in the first Availability Zone
      Type: String
      Default: 10.192.20.0/24
    PrivateSubnet2CIDR:
      Description: The IP range (CIDR notation) for the private subnet in the second Availability Zone
      Type: String
      Default: 10.192.21.0/24
    MaxWorkerNodes:
      Description: The maximum number of workers that can run in the environment
      Type: Number
      Default: 2
    DagProcessingLogs:
      Description: Log level for DagProcessing
      Type: String
      Default: INFO
    SchedulerLogsLevel:
      Description: Log level for SchedulerLogs
      Type: String
      Default: INFO
    TaskLogsLevel:
      Description: Log level for TaskLogs
      Type: String
      Default: INFO
    WorkerLogsLevel:
      Description: Log level for WorkerLogs
      Type: String
      Default: INFO
    WebserverLogsLevel:
      Description: Log level for WebserverLogs
      Type: String
      Default: INFO
  
  Resources:
    #####################################################################################################################
    # CREATE VPC
    #####################################################################################################################
  
    VPC:
      Type: AWS::EC2::VPC
      Properties:
        CidrBlock: !Ref VpcCIDR
        EnableDnsSupport: true
        EnableDnsHostnames: true
        Tags:
          - Key: Name
            Value: MWAAEnvironment
  
    InternetGateway:
      Type: AWS::EC2::InternetGateway
      Properties:
        Tags:
          - Key: Name
            Value: MWAAEnvironment
  
    InternetGatewayAttachment:
      Type: AWS::EC2::VPCGatewayAttachment
      Properties:
        InternetGatewayId: !Ref InternetGateway
        VpcId: !Ref VPC
  
    PublicSubnet1:
      Type: AWS::EC2::Subnet
      Properties:
        VpcId: !Ref VPC
        AvailabilityZone: !Select [ 0, !GetAZs '' ]
        CidrBlock: !Ref PublicSubnet1CIDR
        MapPublicIpOnLaunch: true
        Tags:
          - Key: Name
            Value: !Sub ${EnvironmentName} Public Subnet (AZ1)
  
    PublicSubnet2:
      Type: AWS::EC2::Subnet
      Properties:
        VpcId: !Ref VPC
        AvailabilityZone: !Select [ 1, !GetAZs  '' ]
        CidrBlock: !Ref PublicSubnet2CIDR
        MapPublicIpOnLaunch: true
        Tags:
          - Key: Name
            Value: !Sub ${EnvironmentName} Public Subnet (AZ2)
  
    PrivateSubnet1:
      Type: AWS::EC2::Subnet
      Properties:
        VpcId: !Ref VPC
        AvailabilityZone: !Select [ 0, !GetAZs  '' ]
        CidrBlock: !Ref PrivateSubnet1CIDR
        MapPublicIpOnLaunch: false
        Tags:
          - Key: Name
            Value: !Sub ${EnvironmentName} Private Subnet (AZ1)
  
    PrivateSubnet2:
      Type: AWS::EC2::Subnet
      Properties:
        VpcId: !Ref VPC
        AvailabilityZone: !Select [ 1, !GetAZs  '' ]
        CidrBlock: !Ref PrivateSubnet2CIDR
        MapPublicIpOnLaunch: false
        Tags:
          - Key: Name
            Value: !Sub ${EnvironmentName} Private Subnet (AZ2)
  
    NatGateway1EIP:
      Type: AWS::EC2::EIP
      DependsOn: InternetGatewayAttachment
      Properties:
        Domain: vpc
  
    NatGateway2EIP:
      Type: AWS::EC2::EIP
      DependsOn: InternetGatewayAttachment
      Properties:
        Domain: vpc
  
    NatGateway1:
      Type: AWS::EC2::NatGateway
      Properties:
        AllocationId: !GetAtt NatGateway1EIP.AllocationId
        SubnetId: !Ref PublicSubnet1
  
    NatGateway2:
      Type: AWS::EC2::NatGateway
      Properties:
        AllocationId: !GetAtt NatGateway2EIP.AllocationId
        SubnetId: !Ref PublicSubnet2
  
    PublicRouteTable:
      Type: AWS::EC2::RouteTable
      Properties:
        VpcId: !Ref VPC
        Tags:
          - Key: Name
            Value: !Sub ${EnvironmentName} Public Routes
  
    DefaultPublicRoute:
      Type: AWS::EC2::Route
      DependsOn: InternetGatewayAttachment
      Properties:
        RouteTableId: !Ref PublicRouteTable
        DestinationCidrBlock: 0.0.0.0/0
        GatewayId: !Ref InternetGateway
  
    PublicSubnet1RouteTableAssociation:
      Type: AWS::EC2::SubnetRouteTableAssociation
      Properties:
        RouteTableId: !Ref PublicRouteTable
        SubnetId: !Ref PublicSubnet1
  
    PublicSubnet2RouteTableAssociation:
      Type: AWS::EC2::SubnetRouteTableAssociation
      Properties:
        RouteTableId: !Ref PublicRouteTable
        SubnetId: !Ref PublicSubnet2
  
  
    PrivateRouteTable1:
      Type: AWS::EC2::RouteTable
      Properties:
        VpcId: !Ref VPC
        Tags:
          - Key: Name
            Value: !Sub ${EnvironmentName} Private Routes (AZ1)
  
    DefaultPrivateRoute1:
      Type: AWS::EC2::Route
      Properties:
        RouteTableId: !Ref PrivateRouteTable1
        DestinationCidrBlock: 0.0.0.0/0
        NatGatewayId: !Ref NatGateway1
  
    PrivateSubnet1RouteTableAssociation:
      Type: AWS::EC2::SubnetRouteTableAssociation
      Properties:
        RouteTableId: !Ref PrivateRouteTable1
        SubnetId: !Ref PrivateSubnet1
  
    PrivateRouteTable2:
      Type: AWS::EC2::RouteTable
      Properties:
        VpcId: !Ref VPC
        Tags:
          - Key: Name
            Value: !Sub ${EnvironmentName} Private Routes (AZ2)
  
    DefaultPrivateRoute2:
      Type: AWS::EC2::Route
      Properties:
        RouteTableId: !Ref PrivateRouteTable2
        DestinationCidrBlock: 0.0.0.0/0
        NatGatewayId: !Ref NatGateway2
  
    PrivateSubnet2RouteTableAssociation:
      Type: AWS::EC2::SubnetRouteTableAssociation
      Properties:
        RouteTableId: !Ref PrivateRouteTable2
        SubnetId: !Ref PrivateSubnet2
  
    SecurityGroup:
      Type: AWS::EC2::SecurityGroup
      Properties:
        GroupName: "mwaa-security-group"
        GroupDescription: "Security group with a self-referencing inbound rule."
        VpcId: !Ref VPC
  
    SecurityGroupIngress:
      Type: AWS::EC2::SecurityGroupIngress
      Properties:
        GroupId: !Ref SecurityGroup
        IpProtocol: "-1"
        SourceSecurityGroupId: !Ref SecurityGroup
  
    EnvironmentBucket:
      Type: AWS::S3::Bucket
      Properties:
        VersioningConfiguration:
          Status: Enabled
        PublicAccessBlockConfiguration: 
          BlockPublicAcls: true
          BlockPublicPolicy: true
          IgnorePublicAcls: true
          RestrictPublicBuckets: true
  
    #####################################################################################################################
    # CREATE MWAA
    #####################################################################################################################
  
    MwaaEnvironment:
      Type: AWS::MWAA::Environment
      DependsOn: MwaaExecutionPolicy
      Properties:
        Name: !Sub "${AWS::StackName}-MwaaEnvironment"
        SourceBucketArn: !GetAtt EnvironmentBucket.Arn
        ExecutionRoleArn: !GetAtt MwaaExecutionRole.Arn
        DagS3Path: dags
        NetworkConfiguration:
          SecurityGroupIds:
            - !GetAtt SecurityGroup.GroupId
          SubnetIds:
            - !Ref PrivateSubnet1
            - !Ref PrivateSubnet2
        WebserverAccessMode: PUBLIC_ONLY
        MaxWorkers: !Ref MaxWorkerNodes
        LoggingConfiguration:
          DagProcessingLogs:
            LogLevel: !Ref DagProcessingLogs
            Enabled: true
          SchedulerLogs:
            LogLevel: !Ref SchedulerLogsLevel
            Enabled: true
          TaskLogs:
            LogLevel: !Ref TaskLogsLevel
            Enabled: true
          WorkerLogs:
            LogLevel: !Ref WorkerLogsLevel
            Enabled: true
          WebserverLogs:
            LogLevel: !Ref WebserverLogsLevel
            Enabled: true
    SecurityGroup:
      Type: AWS::EC2::SecurityGroup
      Properties:
        VpcId: !Ref VPC
        GroupDescription: !Sub "Security Group for Amazon MWAA Environment ${AWS::StackName}-MwaaEnvironment"
        GroupName: !Sub "airflow-security-group-${AWS::StackName}-MwaaEnvironment"
    
    SecurityGroupIngress:
      Type: AWS::EC2::SecurityGroupIngress
      Properties:
        GroupId: !Ref SecurityGroup
        IpProtocol: "-1"
        SourceSecurityGroupId: !Ref SecurityGroup
  
    SecurityGroupEgress:
      Type: AWS::EC2::SecurityGroupEgress
      Properties:
        GroupId: !Ref SecurityGroup
        IpProtocol: "-1"
        CidrIp: "0.0.0.0/0"
  
    MwaaExecutionRole:
      Type: AWS::IAM::Role
      Properties:
        AssumeRolePolicyDocument:
          Version: 2012-10-17
          Statement:
            - Effect: Allow
              Principal:
                Service:
                  - airflow-env.amazonaws.com
                  - airflow.amazonaws.com
              Action:
               - "sts:AssumeRole"
        Path: "/service-role/"
  
    MwaaExecutionPolicy:
      DependsOn: EnvironmentBucket
      Type: AWS::IAM::ManagedPolicy
      Properties:
        Roles:
          - !Ref MwaaExecutionRole
        PolicyDocument:
          Version: 2012-10-17
          Statement:
            - Effect: Allow
              Action: airflow:PublishMetrics
              Resource:
                - !Sub "arn:aws:airflow:${AWS::Region}:${AWS::AccountId}:environment/${EnvironmentName}"
            - Effect: Deny
              Action: s3:ListAllMyBuckets
              Resource:
                - !Sub "${EnvironmentBucket.Arn}"
                - !Sub "${EnvironmentBucket.Arn}/*"
  
            - Effect: Allow
              Action:
                - "s3:GetObject*"
                - "s3:GetBucket*"
                - "s3:List*"
              Resource:
                - !Sub "${EnvironmentBucket.Arn}"
                - !Sub "${EnvironmentBucket.Arn}/*"
            - Effect: Allow
              Action:
                - logs:DescribeLogGroups
              Resource: "*"
  
            - Effect: Allow
              Action:
                - logs:CreateLogStream
                - logs:CreateLogGroup
                - logs:PutLogEvents
                - logs:GetLogEvents
                - logs:GetLogRecord
                - logs:GetLogGroupFields
                - logs:GetQueryResults
                - logs:DescribeLogGroups
              Resource:
                - !Sub "arn:aws:logs:${AWS::Region}:${AWS::AccountId}:log-group:airflow-${AWS::StackName}*"
            - Effect: Allow
              Action: cloudwatch:PutMetricData
              Resource: "*"
            - Effect: Allow
              Action:
                - sqs:ChangeMessageVisibility
                - sqs:DeleteMessage
                - sqs:GetQueueAttributes
                - sqs:GetQueueUrl
                - sqs:ReceiveMessage
                - sqs:SendMessage
              Resource:
                - !Sub "arn:aws:sqs:${AWS::Region}:*:airflow-celery-*"
            - Effect: Allow
              Action:
                - kms:Decrypt
                - kms:DescribeKey
                - "kms:GenerateDataKey*"
                - kms:Encrypt
              NotResource: !Sub "arn:aws:kms:*:${AWS::AccountId}:key/*"
              Condition:
                StringLike:
                  "kms:ViaService":
                    - !Sub "sqs.${AWS::Region}.amazonaws.com"
  Outputs:
    VPC:
      Description: A reference to the created VPC
      Value: !Ref VPC
  
    PublicSubnets:
      Description: A list of the public subnets
      Value: !Join [ ",", [ !Ref PublicSubnet1, !Ref PublicSubnet2 ]]
  
    PrivateSubnets:
      Description: A list of the private subnets
      Value: !Join [ ",", [ !Ref PrivateSubnet1, !Ref PrivateSubnet2 ]]
  
    PublicSubnet1:
      Description: A reference to the public subnet in the 1st Availability Zone
      Value: !Ref PublicSubnet1
  
    PublicSubnet2:
      Description: A reference to the public subnet in the 2nd Availability Zone
      Value: !Ref PublicSubnet2
  
    PrivateSubnet1:
      Description: A reference to the private subnet in the 1st Availability Zone
      Value: !Ref PrivateSubnet1
  
    PrivateSubnet2:
      Description: A reference to the private subnet in the 2nd Availability Zone
      Value: !Ref PrivateSubnet2
  
    SecurityGroupIngress:
      Description: Security group with self-referencing inbound rule
      Value: !Ref SecurityGroupIngress
  
    MwaaApacheAirflowUI:
      Description: MWAA Environment
      Value: !Sub  "https://${MwaaEnvironment.WebserverUrl}"
  

第二步：使用 AWS CLI

1. 在命令提示符中，導航到存儲mwaa-public-network.yml的目錄。例如：

   cd mwaaproject

2. 使用aws cloudformation create-stack指令建立使用的堆疊 AWS CLI。

   aws cloudformation create-stack --stack-name mwaa-environment-public-network --template-body file://mwaa-public-network.yml --capabilities CAPABILITY_IAM

   注意

   建立 Amazon VPC 基礎設施、Amazon S3 儲存貯體和亞馬遜 MWAA 環境需要 30 分鐘以上的時間。

第三步：將 DAG 上傳到 Amazon S3 並在 Apache 氣流 UI 中運行

1. 複製最新支援 Apache 氣流版本的tutorial.py檔案內容，並在本機儲存為tutorial.py。

2. 在命令提示符中，導航到存儲tutorial.py的目錄。例如：

   cd mwaaproject

3. 使用下列命令列出所有 Amazon S3 儲存貯體。

   aws s3 ls

4. 使用下列命令列出您環境之 Amazon S3 儲存貯體中的檔案和資料夾。

   aws s3 ls s3://YOUR_S3_BUCKET_NAME

5. 使用下面的腳本將tutorial.py文件上傳到您的文件dags夾。取代「您的 _ S3_ 桶名稱」中的樣本值。

   aws s3 cp tutorial.py s3://YOUR_S3_BUCKET_NAME/dags/

6. 在 Amazon MWAA 主控台上開啟「環境」頁面。

7. 選擇一個環境。

8. 選擇「開啟氣流 UI」。

9. 在 Apache 氣流使用者介面上，從可用的 DAG 清單中選擇教學課程 DAG。

10. 在 DAG 詳細資料頁面上，選擇 DAG 名稱旁的暫停/取消暫停 DAG 切換，以取消暫停 DAG。

11. 選擇 [觸發 DAG]。

步驟四：在記錄檔中檢視 CloudWatch 記錄

您可以在 CloudWatch 主控台中檢視 AWS CloudFormation 堆疊啟用的所有 Apache 氣流記錄檔的 Apache 氣流記錄檔。下節說明如何檢視 Airflow Web 伺服器記錄群組的記錄檔。

1. 在 Amazon MWAA 主控台上開啟「環境」頁面。

2. 選擇一個環境。

3. 在 [監視] 窗格中選擇 Airflow Web 伺服器記錄群組。

4. 在「webserver_console_ip記錄資料流」中選擇記錄檔。

後續步驟？

• 進一步瞭解如何上傳 DAG、在中指定 Python 相依性，以requirements.txt及在中的自訂外掛程式中指定plugins.zip。在亞馬遜 MWAA 與 DAG 工作

• 進一步了解我們建議您調整環境效能的最佳實務Amazon MWAA 上阿帕奇氣流的性能調整。

• 在中為您的環境建立監視儀表板監控 Amazon MWAA 上的儀表板和警報。

• 在中執行一些 DAG 程式碼範例適用於 Apache 氣流的 Amazon 受管工作流程程式。