Las traducciones son generadas a través de traducción automática. En caso de conflicto entre la traducción y la version original de inglés, prevalecerá la version en inglés.

Procesa mensajes de gran volumen de Amazon SQS con los flujos de trabajo de Step Functions Express

En este proyecto de ejemplo se muestra cómo utilizar un AWS Step Functions Flujo de trabajo rápido para procesar mensajes o datos de una fuente de eventos de gran volumen, como Amazon Simple Queue Service (AmazonSQS). Dado que los flujos de trabajo rápidos se pueden iniciar a una velocidad muy elevada, son ideales para las cargas de trabajo de datos de streaming o procesamiento de eventos de un volumen elevado.

A continuación se muestran dos métodos utilizados con frecuencia para ejecutar la máquina de estado desde un origen de eventos:

Este proyecto de ejemplo usa el segundo método para iniciar una ejecución cada vez que la SQS cola de Amazon envía un mensaje. Puede usar una configuración similar para activar la ejecución de flujos de trabajo rápidos desde otros orígenes de eventos, como Amazon Simple Storage Service (Amazon S3), Amazon DynamoDB y Amazon Kinesis.

Para obtener más información acerca de los flujos de trabajo rápidos y las integraciones de servicios de Step Functions, consulte los siguientes temas:

Paso 1: Crear la máquina de estado

Los siguientes pasos dependen de la elección anterior:

Paso 2: Activar la ejecución de la máquina de estado

Código de función de Lambda de ejemplo

A continuación se muestra el código de una función Lambda que muestra cómo la función Lambda iniciadora inicia la ejecución de una máquina de estados mediante AWS SDK.

import boto3

def lambda_handler(event, context):
    message_body = event['Records'][0]['body']
    client = boto3.client('stepfunctions')
    response = client.start_execution(
        stateMachineArn='${ExpressStateMachineArn}',
        input=message_body
    )

Código de la máquina de estado de ejemplo

El flujo de trabajo rápido de este proyecto de ejemplo está compuesto por un conjunto de funciones Lambda para procesar textos.

Para obtener más información sobre cómo AWS Step Functions puede controlar otros AWS servicios, consulteIntegración de servicios con Step Functions.

{
  "Comment": "An example of using Express workflows to run text processing for each message sent from an SQS queue.",
  "StartAt": "Decode base64 string",
  "States": {
    "Decode base64 string": {
      "Type": "Task",
      "Resource": "arn:<PARTITION>:states:::lambda:invoke",
      "OutputPath": "$.Payload",
      "Parameters": {
        "FunctionName": "<BASE64_DECODER_LAMBDA_FUNCTION_NAME>",
        "Payload.$": "$"
      },
      "Next": "Generate statistics"
    },
    "Generate statistics": {
      "Type": "Task",
      "Resource": "arn:<PARTITION>:states:::lambda:invoke",
      "OutputPath": "$.Payload",
      "Parameters": {
        "FunctionName": "<TEXT_STATS_GENERATING_LAMBDA_FUNCTION_NAME>",
        "Payload.$": "$"
      },
      "Next": "Remove special characters"
    },
    "Remove special characters": {
      "Type": "Task",
      "Resource": "arn:<PARTITION>:states:::lambda:invoke",
      "OutputPath": "$.Payload",
      "Parameters": {
        "FunctionName": "<STRING_CLEANING_LAMBDA_FUNCTION_NAME>",
        "Payload.$": "$"
      },
      "Next": "Tokenize and count"
    },
    "Tokenize and count": {
      "Type": "Task",
      "Resource": "arn:<PARTITION>:states:::lambda:invoke",
      "OutputPath": "$.Payload",
      "Parameters": {
        "FunctionName": "<TOKENIZING_AND_WORD_COUNTING_LAMBDA_FUNCTION_NAME>",
        "Payload.$": "$"
      },
      "End": true
    }
  }
}

IAMEjemplo

Este ejemplo AWS Identity and Access Management La política (IAM) generada por el proyecto de muestra incluye el mínimo de privilegios necesarios para ejecutar la máquina de estados y los recursos relacionados. Le recomendamos que incluya solo los permisos que sean necesarios en sus IAM políticas.

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Action": [
                "lambda:InvokeFunction"
            ],
            "Resource": [
                "arn:aws:lambda:us-east-1:123456789012:function:example-Base64DecodeLambda-wJalrXUtnFEMI",
                "arn:aws:lambda:us-east-1:123456789012:function:example-StringCleanerLambda-je7MtGbClwBF",
                "arn:aws:lambda:us-east-1:123456789012:function:example-TokenizerCounterLambda-wJalrXUtnFEMI",
                "arn:aws:lambda:us-east-1:123456789012:function:example-GenerateStatsLambda-je7MtGbClwBF"
            ],
            "Effect": "Allow"
        }
    ]
}

La siguiente política garantiza que haya permisos suficientes para los CloudWatch registros.

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Action": [
                "logs:CreateLogDelivery",
                "logs:GetLogDelivery",
                "logs:UpdateLogDelivery",
                "logs:DeleteLogDelivery",
                "logs:ListLogDeliveries",
                "logs:PutResourcePolicy",
                "logs:DescribeResourcePolicies",
                "logs:DescribeLogGroups"
            ],
            "Resource": [
                "*"
            ],
            "Effect": "Allow"
        }
    ]
}

Para obtener información sobre cómo configurar el uso IAM de Step Functions con otros AWS servicios, consulteCómo Step Functions genera IAM políticas para servicios integrados.