Les traductions sont fournies par des outils de traduction automatique. En cas de conflit entre le contenu d'une traduction et celui de la version originale en anglais, la version anglaise prévaudra.

Traitez de gros volumes de messages provenant d'Amazon SQS avec les flux de travail Step Functions Express

Cet exemple de projet montre comment utiliser un AWS Step Functions Express Workflow pour traiter des messages ou des données provenant d'une source d'événements à volume élevé, telle qu'Amazon Simple Queue Service (AmazonSQS). Comme Express Workflows peut être démarré à un rythme très élevé, il convient parfaitement pour le traitement d'événements à volume élevé ou les charges de travail de données en continu.

Voici deux méthodes couramment utilisées pour exécuter votre machine d'état à partir d'une source d'événement :

Cet exemple de projet utilise la deuxième méthode pour démarrer une exécution chaque fois que la SQS file d'attente Amazon envoie un message. Vous pouvez utiliser une configuration similaire pour déclencher l'exécution d'Express Workflows à partir d'autres sources d'événements, telles qu'Amazon Simple Storage Service (Amazon S3), Amazon DynamoDB et Amazon Kinesis.

Pour plus d'informations sur les intégrations des services Express Workflows et Step Functions, consultez les rubriques suivantes :

Étape 1 : Création de la machine à états

Les prochaines étapes dépendent de votre choix précédent :

Étape 2 : Déclencher l'exécution de la machine à états

Exemple de code de fonction Lambda

Le code de fonction Lambda ci-dessous montre comment la fonction Lambda initiatrice démarre une exécution par machine à états à l'aide du 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
    )

Exemple de code de machine d'état

Le workflow express de cet exemple de projet consiste en un ensemble de fonctions Lambda pour le traitement de texte.

Pour plus d'informations sur la façon dont AWS Step Functions peut contrôler d'autres AWS services, voirIntégrer les services avec 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
    }
  }
}

IAMExemple

Cet exemple AWS Identity and Access Management (IAM) la politique générée par l'exemple de projet inclut le moindre privilège nécessaire pour exécuter la machine à états et les ressources associées. Nous vous recommandons de n'inclure que les autorisations nécessaires dans vos IAM politiques.

{
    "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 politique suivante garantit que les autorisations sont suffisantes pour les CloudWatch journaux.

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

Pour plus d'informations sur la configuration IAM lors de l'utilisation de Step Functions avec d'autres AWS services, voirComment Step Functions génère IAM des politiques pour les services intégrés.