Tutoriais de início rápido para Amazon Managed Workflows for Apache Airflow

Este tutorial de início rápido usa um AWS CloudFormation modelo que cria a infraestrutura Amazon VPC, um bucket Amazon S3 com uma pasta e dags um ambiente Amazon Managed Workflows for Apache Airflow ao mesmo tempo.

Neste tutorial

Este tutorial mostra três AWS Command Line Interface (AWS CLI) comandos para fazer upload de um DAG para o Amazon S3, executar o DAG no Apache Airflow e visualizar os logins. CloudWatch Conclui mostrando as etapas para criar uma política do IAM para uma equipe de desenvolvimento do Apache Airflow.

nota

O AWS CloudFormation modelo nesta página cria um ambiente Amazon Managed Workflows for Apache Airflow para a versão mais recente do Apache Airflow disponível em. AWS CloudFormation A versão mais recente disponível é o Apache Airflow v2.8.1.

O AWS CloudFormation modelo nesta página cria o seguinte:

• Infraestrutura da VPC. O modelo usa Roteamento público pela Internet. Usa o Modo de acesso à rede pública para o servidor Web do Apache Airflow em WebserverAccessMode: PUBLIC_ONLY.

• Bucket do Amazon S3. O modelo cria um bucket do Amazon S3 com uma pasta dags. Ele está configurado para bloquear todo o acesso público, com o Versionamento de bucket habilitado, conforme definido em Criar um bucket do Amazon S3 para o Amazon MWAA.

• Ambiente do Amazon MWAA. O modelo cria um ambiente Amazon MWAA associado à dags pasta no bucket do Amazon S3, uma função de execução com permissão AWS para serviços usados pelo Amazon MWAA e o padrão para criptografia usando AWS uma chave própria, conforme definido em. Criar um ambiente do Amazon MWAA

• CloudWatch Registros. O modelo permite que o Apache Airflow faça login CloudWatch no nível “INFO” ou superior para o grupo de registros do agendador do Airflow, o grupo de registros do servidor web do Airflow, o grupo de registros de trabalho do Airflow, o grupo de registros de processamento do Airflow DAG e o grupo de registros de tarefas do Airflow, conforme definido em. Visualizando registros de fluxo de ar na Amazon CloudWatch

Você concluirá as seguintes etapas neste tutorial:

• Fazer upload e executar um DAG. Faça o upload do tutorial DAG do Apache Airflow para a versão mais recente do Apache Airflow compatível com o Amazon MWAA para o Amazon S3 e, em seguida, execute na IU do Apache Airflow, conforme definido em Como adicionar ou atualizar DAGs.

• Visualizar logs. Visualize o grupo de registros do servidor web Airflow em CloudWatch Logs, conforme definido emVisualizando registros de fluxo de ar na Amazon CloudWatch.

• Criar uma política de controle de acesso. Crie uma política de controle de acesso no IAM para sua equipe de desenvolvimento do Apache Airflow, conforme definido em Como acessar um ambiente do Amazon MWAA.

Pré-requisitos

O AWS Command Line Interface (AWS CLI) é uma ferramenta de código aberto que permite interagir com AWS serviços usando comandos em seu shell de linha de comando. Para concluir as etapas nesta página, é necessário o seguinte:

• AWS CLI — Instale a versão 2.

• AWS CLI — Configuração rápida com aws configure.

Etapa 1: salvar o AWS CloudFormation modelo localmente

• Copie o conteúdo do modelo a seguir e salve localmente como mwaa-public-network.yml. Também é possível baixar o modelo.

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

Etapa 2: criar a pilha usando o AWS CLI

1. No prompt de comando, navegue até o diretório em que mwaa-public-network.yml está armazenado. Por exemplo: .

   cd mwaaproject

2. Use o comando aws cloudformation create-stack para criar a pilha usando o AWS CLI.

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

   nota

   São necessários mais de 30 minutos para criar a infraestrutura Amazon VPC, o bucket do Amazon S3 e o ambiente do Amazon MWAA.

Etapa 3: faça o upload de um DAG para o Amazon S3 e execute-o na IU do Apache Airflow

1. Copie o conteúdo do arquivo tutorial.py para a versão mais recente compatível do Apache Airflow e salve localmente como tutorial.py.

2. No prompt de comando, navegue até o diretório em que tutorial.py está armazenado. Por exemplo: .

   cd mwaaproject

3. Use o comando a seguir para listar todos os seus buckets do Amazon S3.

   aws s3 ls

4. Use o seguinte comando para listar os arquivos e pastas no bucket do Amazon S3 para seu ambiente.

   aws s3 ls s3://YOUR_S3_BUCKET_NAME

5. Use o script a seguir para fazer upload do arquivo tutorial.py para sua pasta dags. Substitua o valor da amostra em YOUR_S3_BUCKET_NAME.

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

6. Abra a página Ambientes no console do Amazon MWAA.

7. Escolha um ambiente.

8. Escolha Abrir a IU do Airflow.

9. Na IU do Apache Airflow, na lista de DAGs disponíveis, escolha o tutorial DAG.

10. Na página de detalhes do DAG, escolha o botão Pausar/Reiniciar o DAG ao lado do nome do DAG para retomar o DAG.

11. Escolha Acionar DAG.

Etapa quatro: Exibir registros em CloudWatch Registros

Você pode visualizar os registros do Apache Airflow no CloudWatch console para todos os registros do Apache Airflow que foram habilitados pela pilha. AWS CloudFormation A seção a seguir mostra como visualizar os logs do grupo de logs do servidor web Airflow.

1. Abra a página Ambientes no console do Amazon MWAA.

2. Escolha um ambiente.

3. Escolha Grupo de logs do servidor web no Airflow no painel Monitoramento.

4. Escolha o log webserver_console_ip em Fluxos de logs.

Próximas etapas

• Saiba mais sobre como fazer upload de DAGs, especificar dependências do Python em um requirements.txt e plug-ins personalizados em um plugins.zip em Como trabalhar com DAGs no Amazon MWAA.

• Saiba mais sobre as melhores práticas que recomendamos para ajustar o desempenho do seu ambiente em Ajuste de desempenho para o Apache Airflow no Amazon MWAA.

• Crie um painel de monitoramento para seu ambiente em Painéis de monitoramento e alarmes no Amazon MWAA.

• Execute alguns exemplos de código do DAG em Exemplos de código para o Amazon Managed Workflows for Apache Airflow.