View a markdown version of this page

Traitement de données par lots avec une fonction Lambda dans Step Functions - AWS Step Functions
Étape 1 : Création de la machine à étatsÉtape 2 : Création de la fonction LambdaÉtape 3 : Exécutez la machine d'état

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.

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 entier d'éléments dans une fonction Lambda. Chaque lot contient un maximum de trois articles. L'état Distributed Map démarre quatre exécutions de flux de travail enfants, chaque exécution traitant trois éléments, tandis qu'une exécution traite un seul élément. Chaque exécution de flux de travail enfant invoque une fonction Lambda qui effectue une itération 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 soit [1, 2, 3, 4, 5, 6, 7, 8, 9, 10] et que le facteur de multiplication soit7. 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 Étape 1 : Création du prototype de flux de travail à états, consultez le didacticiel Getting started with 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 effectue une itération 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 transmis 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:region:account-id: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 inférieure à Région AWS celle de votre 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, consultez Étape 4 : Configuration de la fonction Lambda dans le didacticiel Getting started with 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é. Voici un exemple d'ARN, où function-name est le nom de la fonction Lambda (dans ce cas,ProcessEntireBatch) :

    arn:aws:lambda:region:123456789012:function:function-name

    Vous devez fournir la fonction ARN dans la machine d'état 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 Distributed Map démarre quatre exécutions de flux de travail enfants, dans lesquelles 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 ProcessEntireBatch fonction par l'une des exécutions du flux de travail enfant.

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

À partir de cette entrée, l'exemple suivant montre le tableau de sortie nommé multiplied qui est renvoyé par la fonction Lambda.

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

La machine à états renvoie la sortie suivante qui contient quatre tableaux nommés multiplied pour 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 Distributed Map pour rechercher tous les multiplied tableaux, extraire tous les éléments de ces tableaux, puis les combiner en un seul tableau de sortie.

Pour utiliser ce 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]
}