Verarbeitung einzelner Elemente mit einer Lambda-Funktion in Step Functions - AWS Step Functions
Schritt 1: Erstellen Sie die ZustandsmaschineSchritt 2: Erstellen Sie die Lambda-FunktionSchritt 3: Führen Sie die Zustandsmaschine aus

Die vorliegende Übersetzung wurde maschinell erstellt. Im Falle eines Konflikts oder eines Widerspruchs zwischen dieser übersetzten Fassung und der englischen Fassung (einschließlich infolge von Verzögerungen bei der Übersetzung) ist die englische Fassung maßgeblich.

Verarbeitung einzelner Elemente mit einer Lambda-Funktion in Step Functions

In diesem Tutorial verwenden Sie das ItemBatcher (Karte) Feld des Distributed Map-Status, um mithilfe einer Lambda-Funktion über einzelne Elemente in einem Batch zu iterieren. Mit dem Status Distributed Map werden vier untergeordnete Workflow-Ausführungen gestartet. Jeder dieser untergeordneten Workflows führt einen Inline-Map-Status aus. Für jede Iteration ruft der Inline-Map-Status eine Lambda-Funktion auf und übergibt ein einzelnes Element aus dem Batch an die Funktion. Die Lambda-Funktion verarbeitet dann das Element und gibt das Ergebnis zurück.

Sie erstellen eine Zustandsmaschine, die eine Multiplikation mit einem Array von ganzen Zahlen durchführt. Angenommen, das Integer-Array, das Sie als Eingabe angeben, ist [1, 2, 3, 4, 5, 6, 7, 8, 9, 10] und der Multiplikationsfaktor ist. 7 Dann wird das resultierende Array, das nach der Multiplikation dieser ganzen Zahlen mit dem Faktor 7 gebildet wird, sein. [7, 14, 21, 28, 35, 42, 49, 56, 63, 70]

Schritt 1: Erstellen Sie die Zustandsmaschine

In diesem Schritt erstellen Sie den Workflow-Prototyp der Zustandsmaschine, die bei jedem Aufruf der Lambda-Funktion, die Sie in Schritt 2 erstellen, ein einzelnes Element aus einem Stapel von Elementen übergibt.

  • Verwenden Sie die folgende Definition, um mithilfe der Step Functions Functions-Konsole eine Zustandsmaschine zu erstellen. Informationen zum Erstellen einer Zustandsmaschine finden Sie Schritt 1: Erstellen Sie den Workflow-Prototyp im Tutorial Erste Schritte mit der Verwendung von Distributed Map State.

    In diesem Zustandsmaschine definieren Sie einen Distributed Map-Status, der ein Array von 10 Ganzzahlen als Eingabe akzeptiert und diese Array-Elemente stapelweise an die untergeordneten Workflow-Ausführungen weitergibt. Jede untergeordnete Workflow-Ausführung erhält einen Stapel mit drei Elementen als Eingabe und führt einen Inline-Map-Status aus. Jede Iteration des Inline-Map-Zustands ruft eine Lambda-Funktion auf und übergibt ein Element aus dem Batch an die Funktion. Diese Funktion multipliziert dann das Element mit dem Faktor von 7 und gibt das Ergebnis zurück.

    Die Ausgabe jeder untergeordneten Workflow-Ausführung ist ein JSON Array, das das Multiplikationsergebnis für jedes der übergebenen Elemente enthält.

    Wichtig

    Achten Sie darauf, den Amazon-Ressourcennamen (ARN) der Lambda-Funktion im folgenden Code durch den Namen ARN der Funktion zu ersetzen, die Sie in Schritt 2 erstellen werden.

    {
  "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": "InnerMap",
        "States": {
          "InnerMap": {
            "Type": "Map",
            "ItemProcessor": {
              "ProcessorConfig": {
                "Mode": "INLINE"
              },
              "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,
            "ItemsPath": "$.Items",
            "ItemSelector": {
              "MyMultiplicationFactor.$": "$.BatchInput.MyMultiplicationFactor",
              "MyItem.$": "$$.Map.Item.Value"
            }
          }
        }
      },
      "End": true,
      "Label": "Map",
      "MaxConcurrency": 1000,
      "ItemsPath": "$.MyItems",
      "ItemBatcher": {
        "MaxItemsPerBatch": 3,
        "BatchInput": {
          "MyMultiplicationFactor.$": "$.MyMultiplicationFactor"
        }
      }
    }
  }
}

Schritt 2: Erstellen Sie die Lambda-Funktion

In diesem Schritt erstellen Sie die Lambda-Funktion, die jedes aus dem Stapel übergebene Element verarbeitet.

Wichtig

Stellen Sie sicher, dass Ihre Lambda-Funktion unter derselben steht AWS-Region als Ihre Zustandsmaschine.

So erstellen Sie die Lambda-Funktion:

  1. Verwenden Sie die Lambda-Konsole, um eine Python-Lambda-Funktion mit dem Namen zu erstellen. ProcessSingleItem Informationen zum Erstellen einer Lambda-Funktion finden Sie unter Schritt 4: Lambda-Funktion konfigurieren im Tutorial Erste Schritte mit der Verwendung von Distributed Map state.

  2. Kopieren Sie den folgenden Code für die Lambda-Funktion und fügen Sie ihn in den Abschnitt Codequelle Ihrer Lambda-Funktion ein.

    import json

def lambda_handler(event, context):
    
    multiplication_factor = event['MyMultiplicationFactor']
    item = event['MyItem']
    
    result = multiplication_factor * item
    
    return {
        'statusCode': 200,
        'multiplied': result
    }

  3. Nachdem Sie Ihre Lambda-Funktion erstellt haben, kopieren Sie die in der oberen rechten Ecke der Seite ARN angezeigte Funktion. Das Folgende ist ein Beispiel, wo ARN Funktionsname ist der Name der Lambda-Funktion (in diesem FallProcessSingleItem):

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

    Sie müssen die Funktion ARN in der Zustandsmaschine bereitstellen, die Sie in Schritt 1 erstellt haben.

  4. Wählen Sie Deploy, um die Änderungen bereitzustellen.

Schritt 3: Führen Sie die Zustandsmaschine aus

Wenn Sie den Zustandsmaschine ausführen, startet der Status Distributed Map vier untergeordnete Workflow-Ausführungen, wobei jede Ausführung drei Elemente verarbeitet, während eine Ausführung ein einzelnes Element verarbeitet.

Das folgende Beispiel zeigt die Daten, die an einen der ProcessSingleItemFunktionsaufrufe innerhalb einer untergeordneten Workflow-Ausführung übergeben werden.

{
  "MyMultiplicationFactor": 7,
  "MyItem": 1
}

Angesichts dieser Eingabe zeigt das folgende Beispiel die Ausgabe, die von der Lambda-Funktion zurückgegeben wird.

{
  "statusCode": 200,
  "multiplied": 7
}

Das folgende Beispiel zeigt das JSON Ausgabearray für eine der untergeordneten Workflow-Ausführungen.

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

Die Zustandsmaschine gibt die folgende Ausgabe zurück, die vier Arrays für die vier untergeordneten Workflow-Ausführungen enthält. Diese Arrays enthalten die Multiplikationsergebnisse der einzelnen Eingabeelemente.

Schließlich ist die State-Machine-Ausgabe ein Array mit dem Namenmultiplied, das alle Multiplikationsergebnisse kombiniert, die für die vier untergeordneten Workflow-Ausführungen zurückgegeben wurden.

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

Um alle Multiplikationsergebnisse, die von den untergeordneten Workflow-Ausführungen zurückgegeben wurden, in einem einzigen Ausgabearray zu kombinieren, können Sie das Feld verwenden. ResultSelector Definieren Sie dieses Feld im Status Distributed Map, um alle Ergebnisse zu finden, die einzelnen Ergebnisse zu extrahieren und sie dann zu einem einzigen Ausgabearray mit dem Namen zu kombinieren. multiplied

Um das ResultSelector Feld zu verwenden, aktualisieren Sie Ihre State-Machine-Definition wie im folgenden Beispiel gezeigt.

{
  "StartAt": "Pass",
  "States": {
    ...
    ...
    "Map": {
      "Type": "Map",
      ...
      ...
      "ItemBatcher": {
        "MaxItemsPerBatch": 3,
        "BatchInput": {
          "MyMultiplicationFactor.$": "$.MyMultiplicationFactor"
        }
      },
      "ItemsPath": "$.MyItems",
      "ResultSelector": {
        "multiplied.$": "$..multiplied"
      }
    }
  }
}

Die aktualisierte Zustandsmaschine gibt ein konsolidiertes Ausgabe-Array zurück, wie im folgenden Beispiel gezeigt.

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