Traitement de données par lots avec une fonction Lambda dans Step Functions

Dans ce didacticiel, vous allez utiliser le ItemBatcher (Carte) champ d'état de la carte distribuée pour traiter un lot complet d'éléments dans une fonction Lambda. Chaque lot contient un maximum de trois articles. L'état de la carte distribuée lance quatre exécutions de flux de travail enfants, où chaque exécution traite trois éléments, tandis qu'une exécution traite un seul élément. Chaque exécution d'un flux de travail enfant invoque une fonction Lambda qui itère sur les éléments individuels présents dans le lot.

Vous allez créer une machine à états qui effectue la multiplication sur un tableau d'entiers. Supposons que le tableau d'entiers que vous fournissez en entrée est [1, 2, 3, 4, 5, 6, 7, 8, 9, 10] et que le facteur de multiplication est7. Ensuite, le tableau résultant formé après avoir multiplié ces entiers par un facteur de 7 sera. [7, 14, 21, 28, 35, 42, 49, 56, 63, 70]

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

Au cours de cette étape, vous créez le prototype de flux de travail de la machine à états qui transmet un lot complet de données à la fonction Lambda que vous allez créer à l'étape 2.

• Utilisez la définition suivante pour créer une machine à états à l'aide de la console Step Functions. Pour plus d'informations sur la création d'une machine à états, consultez Étape 1 : Création du prototype de flux de travail le didacticiel Getting started with using Distributed Map state.

  Dans cette machine à états, vous définissez un état de carte distribuée qui accepte un tableau de 10 entiers en entrée et transmet ce tableau à une fonction Lambda par lots de. 3 La fonction Lambda itère sur les éléments individuels présents dans le lot et renvoie un tableau de sortie nommé. multiplied Le tableau de sortie contient le résultat de la multiplication effectuée sur les éléments passés dans le tableau d'entrée.

  Important

  Assurez-vous de remplacer l'Amazon Resource Name (ARN) de la fonction Lambda dans le code suivant par l'ARN de la fonction que vous allez créer à l'étape 2.

  {
    "StartAt": "Pass",
    "States": {
      "Pass": {
        "Type": "Pass",
        "Next": "Map",
        "Result": {
          "MyMultiplicationFactor": 7,
          "MyItems": [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
        }
      },
      "Map": {
        "Type": "Map",
        "ItemProcessor": {
          "ProcessorConfig": {
            "Mode": "DISTRIBUTED",
            "ExecutionType": "STANDARD"
          },
          "StartAt": "Lambda Invoke",
          "States": {
            "Lambda Invoke": {
              "Type": "Task",
              "Resource": "arn:aws:states:::lambda:invoke",
              "OutputPath": "$.Payload",
              "Parameters": {
                "Payload.$": "$",
                "FunctionName": "arn:aws:lambda:us-east-1:123456789012:function:functionName"
              },
              "Retry": [
                {
                  "ErrorEquals": [
                    "Lambda.ServiceException",
                    "Lambda.AWSLambdaException",
                    "Lambda.SdkClientException",
                    "Lambda.TooManyRequestsException"
                  ],
                  "IntervalSeconds": 2,
                  "MaxAttempts": 6,
                  "BackoffRate": 2
                }
              ],
              "End": true
            }
          }
        },
        "End": true,
        "Label": "Map",
        "MaxConcurrency": 1000,
        "ItemBatcher": {
          "MaxItemsPerBatch": 3,
          "BatchInput": {
            "MyMultiplicationFactor.$": "$.MyMultiplicationFactor"
          }
        },
        "ItemsPath": "$.MyItems"
      }
    }
  }

Étape 2 : Création de la fonction Lambda

Au cours de cette étape, vous créez la fonction Lambda qui traite tous les éléments transmis dans le lot.

Important

Assurez-vous que votre fonction Lambda est identique à celle de votre Région AWS machine à états.

Pour créer la fonction Lambda

1. Utilisez la console Lambda pour créer une fonction Lambda Python nommée. ProcessEntireBatch Pour plus d'informations sur la création d'une fonction Lambda, voir Étape 4 : Configuration de la fonction Lambda dans le didacticiel Getting started with using Distributed Map state.

2. Copiez le code suivant pour la fonction Lambda et collez-le dans la section Code source de votre fonction Lambda.

   import json
   
   def lambda_handler(event, context):
       multiplication_factor = event['BatchInput']['MyMultiplicationFactor']
       items = event['Items']
       
       results = [multiplication_factor * item for item in items]
       
       return {
           'statusCode': 200,
           'multiplied': results
       }

3. Après avoir créé votre fonction Lambda, copiez l'ARN de la fonction affiché dans le coin supérieur droit de la page. Voici un exemple d'ARN, où function-name est le nom de la fonction Lambda (dans ce cas,ProcessEntireBatch) :

   arn:aws:lambda:us-east-1:123456789012:function:function-name

   Vous devrez fournir la fonction ARN dans la machine à états que vous avez créée à l'étape 1.

4. Choisissez Déployer pour déployer les modifications.

Étape 3 : Exécutez la machine d'état

Lorsque vous exécutez la machine à états, l'état de la carte distribuée lance quatre exécutions de flux de travail enfants, où chaque exécution traite trois éléments, tandis qu'une exécution traite un seul élément.

L'exemple suivant montre les données transmises à la ProcessEntireBatchfonction par l'une des exécutions du flux de travail enfant.

{
  "BatchInput": {
    "MyMultiplicationFactor": 7
  },
  "Items": [1, 2, 3]
}

Compte tenu de cette entrée, l'exemple suivant montre le tableau de sortie nommé multiplied renvoyé par la fonction Lambda.

{
  "statusCode": 200,
  "multiplied": [7, 14, 21]
}

La machine d'état renvoie la sortie suivante qui contient quatre tableaux nommés d'après multiplied les quatre exécutions de flux de travail enfants. Ces tableaux contiennent les résultats de multiplication des éléments d'entrée individuels.

[
  {
    "statusCode": 200,
    "multiplied": [7, 14, 21]
  },
  {
    "statusCode": 200,
    "multiplied": [28, 35, 42]
  },
  {
    "statusCode": 200,
    "multiplied": [49, 56, 63]
  },
  {
    "statusCode": 200,
    "multiplied": [70]
  }
]

Pour combiner tous les éléments du tableau renvoyés dans un seul tableau de sortie, vous pouvez utiliser le ResultSelector champ. Définissez ce champ dans l'état de la carte distribuée pour rechercher tous les multiplied tableaux, extraire tous les éléments contenus dans ces tableaux, puis les combiner en un seul tableau de sortie.

Pour utiliser le ResultSelector champ, mettez à jour la définition de votre machine à états comme indiqué dans l'exemple suivant.

{
  "StartAt": "Pass",
  "States": {
    ...
    ...
    "Map": {
      "Type": "Map",
      ...
      ...
      "ItemsPath": "$.MyItems",
      "ResultSelector": {
        "multiplied.$": "$..multiplied[*]"
      }
    }
  }
}

La machine d'état mise à jour renvoie un tableau de sortie consolidé, comme indiqué dans l'exemple suivant.

{
  "multiplied": [7, 14, 21, 28, 35, 42, 49, 56, 63, 70]
}