Amazon Managed Workflows for Apache Airflow용 빠른 시작 튜토리얼

이 퀵 스타트 자습서에서는 Amazon VPC 인프라, dags 폴더가 있는 Amazon S3 버킷, Apache Airflow용 Amazon 관리형 워크플로를 동시에 생성하는 AWS CloudFormation 템플릿을 사용합니다.

이 튜토리얼에서는

이 자습서에서는 Amazon S3에 DAG를 업로드하고, Apache Airflow에서 DAG를 실행하고, 로그인을 확인하는 세 가지 AWS Command Line Interface (AWS CLI) 명령을 안내합니다. CloudWatch 마지막으로 Apache Airflow 개발 팀을 위한 IAM 정책을 생성하는 단계를 안내합니다.

참고

이 페이지의 AWS CloudFormation 템플릿은 에서 사용 가능한 최신 버전의 Apache Airflow를 위한 Amazon Apache Airflow용 관리형 워크플로 환경을 생성합니다. AWS CloudFormation사용 가능한 최신 버전은 아파치 에어플로우 v2.8.1입니다.

이 페이지의 AWS CloudFormation 템플릿은 다음을 생성합니다.

• VPC 인프라. 이 템플릿은 인터넷을 통한 퍼블릭 라우팅을(를) 사용합니다. WebserverAccessMode: PUBLIC_ONLY의 Apache Airflow 웹 서버에 퍼블릭 네트워크 액세스 모드을(를) 사용합니다.

• Amazon S3 버킷. 템플릿은 dags 폴더가 있는 Amazon S3 버킷을 생성합니다. Amazon MWAA용 Amazon S3 버킷 생성에 정의된 대로 버킷 버저닝을 활성화하여 모든 퍼블릭 액세스를 차단하도록 구성되어 있습니다.

• Amazon MWAA 환경. 템플릿은 Amazon S3 버킷의 dags 폴더와 연결된 Amazon MWAA 환경, Amazon MWAA에서 사용하는 AWS 서비스에 대한 권한을 가진 실행 역할, 그리고 에서 정의한 대로 AWS 소유 키를 사용한 암호화의 기본 환경을 생성합니다. Amazon MWAA 환경 생성

• CloudWatch 로그. 템플릿을 사용하면 에 정의된 CloudWatch 대로 Apache Airflow를 Airflow 스케줄러 로그 그룹, Airflow 웹 서버 로그 그룹, Airflow 작업자 로그 그룹, Airflow DAG 처리 로그 그룹 및 Airflow 작업 로그 그룹에 대해 “정보” 수준 이상으로 로그인할 수 있습니다. Amazon에서 에어플로우 로그 보기 CloudWatch

이 튜토리얼에서는 다음 작업을 수행할 수 있습니다.

• DAG를 업로드하고 실행. Amazon MWAA가 지원하는 최신 Apache Airflow 버전에 대한 Apache Airflow 튜토리얼 DAG를 Amazon S3에 업로드한 다음 DAG 추가 또는 업데이트에 정의된 대로 Apache Airflow UI에서 실행할 수 있습니다.

• 로그 확인. 에 정의된 대로 로그에서 Airflow 웹 서버 로그 그룹을 확인하십시오. CloudWatch Amazon에서 에어플로우 로그 보기 CloudWatch

• 액세스 제어 정책 생성. Amazon MWAA 환경 액세스에 정의된 대로 Apache Airflow 개발 팀을 위한 액세스 제어 정책을 IAM에서 생성합니다.

필수 조건

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}"
  

2단계: 를 사용하여 스택 생성 AWS CLI

1. 명령 프롬프트에서 mwaa-public-network.yml이 저장된 디렉터리로 이동합니다. 예:

   cd mwaaproject

2. AWS CLI을(를) 사용하여 스택을 생성하려면 aws cloudformation create-stack 명령을 사용합니다.

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

   참고

   Amazon VPC 인프라, Amazon S3 버킷, Amazon MWAA 환경을 생성하는 데 30분 이상이 소요됩니다.

3단계: Amazon S3에 DAG를 업로드하고 Apache Airflow UI에서 실행

1. 지원되는 최신 Apache Airflow 버전의 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 폴더에 업로드할 수 있습니다. YOUR_S3_BUCKET_NAME의 샘플 값을 대체합니다.

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

6. Amazon MWAA 콘솔에서 환경 페이지를 엽니다.

7. 환경을 선택합니다.

8. Airflow UI 열기를 선택합니다.

9. Apache Airflow UI의 사용 가능한 DAG 목록에서 튜토리얼 DAG를 선택합니다.

10. DAG 세부 정보 페이지에서 DAG 이름 옆에 있는 DAG 일시 중지/일시 중지 해제 토글을 선택하여 DAG 일시 중지를 해제합니다.

11. 트리거 DAG를 선택합니다.

4단계: 로그에서 CloudWatch 로그 보기

CloudWatch 콘솔에서 스택에 의해 활성화된 모든 Apache Airflow 로그의 Apache Airflow 로그를 볼 수 있습니다. AWS CloudFormation 다음 섹션에서는 Airflow 웹 서버 로그 그룹의 로그를 보는 방법을 확인할 수 있습니다.

1. Amazon MWAA 콘솔에서 환경 페이지를 엽니다.

2. 환경을 선택합니다.

3. 모니터링 창에서 Airflow 웹 서버 로그 그룹을 선택합니다.

4. 로그 스트림의 webserver_console_ip 로그를 선택합니다.

다음 단계

• DAG를 업로드하고, requirements.txt에서 Python 종속성을 지정하고, Amazon MWAA에서 DAG 작업의 plugins.zip에서 사용자 지정 플러그인을 지정하는 방법에 대해 자세히 알아봅니다.

• Amazon MWAA의 Apache Airflow 성능 튜닝에서 환경 성능 조정을 위해 권장하는 모범 사례에 대해 자세히 알아봅니다.

• Amazon MWAA의 대시보드 및 알람 모니터링에서 사용자 환경을 위한 모니터링 대시보드를 생성합니다.

• Amazon Managed Workflows for Apache Airflow용 코드 예제에서 일부 DAG 코드 샘플을 실행합니다.