Dynamisches Verarbeiten von Daten mit einem Map-Status

Dieses Beispielprojekt demonstriert die dynamische Parallelität mithilfe eines Zuordnung-Zustands.

In diesem Projekt verwendet Step Functions eine AWS Lambda Funktion, um Nachrichten aus einer Amazon SQS SQS-Warteschlange abzurufen und ein JSON-Array dieser Nachrichten an einen Map Status weiterzuleiten. Für jede Nachricht in der Warteschlange schreibt die Zustandsmaschine die Nachricht in DynamoDB, ruft die andere Lambda-Funktion auf, um die Nachricht aus Amazon SQS zu entfernen, und veröffentlicht die Nachricht dann im Amazon SNS SNS-Thema.

Weitere Informationen zu Map Status- und Step Functions Functions-Dienstintegrationen finden Sie im Folgenden:

Schritt 1: Erstellen Sie die Zustandsmaschine und stellen Sie Ressourcen bereit

Nachdem die Ressourcen des Beispielprojekts bereitgestellt wurden, müssen Sie Elemente zur Amazon SQS-Warteschlange hinzufügen und das Amazon SNS-Thema abonnieren, bevor Sie die Zustandsmaschine ausführen.

Schritt 2: Abonnieren Sie das Amazon SNS SNS-Thema

Schritt 3: Nachrichten zur Amazon SQS SQS-Warteschlange hinzufügen

Schritt 4: Führen Sie die Zustandsmaschine aus

Beispiel für einen State-Machine-Code

Die Zustandsmaschine in diesem Beispielprojekt lässt sich in Amazon SQS, Amazon SNS und Lambda integrieren, indem Parameter direkt an diese Ressourcen übergeben werden.

Sehen Sie sich diese Beispiel-Zustandsmaschine an, um zu sehen, wie Step Functions Lambda, DynamoDB und Amazon SNS steuert, indem es eine Verbindung zum Amazon-Ressourcennamen (ARN) im Resource Feld herstellt und Parameters an die Service-API weiterleitet.

Weitere Informationen darüber, wie Sie andere AWS Dienste steuern AWS Step Functions können, finden Sie unter. Verwendung AWS Step Functions mit anderen Diensten

{
  "Comment": "An example of the Amazon States Language for reading messages from an SQS queue and iteratively processing each message.",
  "StartAt": "Read messages from SQS Queue",
  "States": {
    "Read messages from SQS Queue": {
      "Type": "Task",
      "Resource": "arn:aws:states:::lambda:invoke",
      "OutputPath": "$.Payload",
      "Parameters": {
        "FunctionName": "MapSampleProj-ReadFromSQSQueueLambda-1MY3M63RMJVA9"
      },
      "Next": "Are there messages to process?"
    },
    "Are there messages to process?": {
      "Type": "Choice",
      "Choices": [
        {
          "Variable": "$",
          "StringEquals": "No messages",
          "Next": "Finish"
        }
      ],
      "Default": "Process messages"
    },
    "Process messages": {
      "Type": "Map",
      "Next": "Finish",
      "ItemsPath": "$",
      "Parameters": {
        "MessageNumber.$": "$$.Map.Item.Index",
        "MessageDetails.$": "$$.Map.Item.Value"
      },
      "Iterator": {
        "StartAt": "Write message to DynamoDB",
        "States": {
          "Write message to DynamoDB": {
            "Type": "Task",
            "Resource": "arn:aws:states:::dynamodb:putItem",
            "ResultPath": null,
            "Parameters": {
              "TableName": "MapSampleProj-DDBTable-YJDJ1MKIN6C5",
              "ReturnConsumedCapacity": "TOTAL",
              "Item": {
                "MessageId": {
                  "S.$": "$.MessageDetails.MessageId"
                },
                "Body": {
                  "S.$": "$.MessageDetails.Body"
                }
              }
            },
            "Next": "Remove message from SQS queue"
          },
          "Remove message from SQS queue": {
            "Type": "Task",
            "Resource": "arn:aws:states:::lambda:invoke",
            "InputPath": "$.MessageDetails",
            "ResultPath": null,
            "Parameters": {
              "FunctionName": "MapSampleProj-DeleteFromSQSQueueLambda-198J2839ZO5K2",
              "Payload": {
                "ReceiptHandle.$": "$.ReceiptHandle"
              }
            },
            "Next": "Publish message to SNS topic"
          },
          "Publish message to SNS topic": {
            "Type": "Task",
            "Resource": "arn:aws:states:::sns:publish",
            "InputPath": "$.MessageDetails",
            "Parameters": {
              "Subject": "Message from Step Functions!",
              "Message.$": "$.Body",
              "TopicArn": "arn:aws:sns:us-east-1:012345678910:MapSampleProj-SNSTopic-1CQO4HQ3IR1KN"
            },
            "End": true
          }
        }
      }
    },
    "Finish": {
      "Type": "Succeed"
    }
  }
}

IAM-Beispiel

Diese vom Beispielprojekt generierte Beispielrichtlinie AWS Identity and Access Management (IAM) beinhaltet die geringsten Rechte, die für die Ausführung der Zustandsmaschine und der zugehörigen Ressourcen erforderlich sind. Wir empfehlen, dass Sie nur die Berechtigungen in Ihre IAM-Richtlinien aufnehmen, die erforderlich sind.

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Action": [
                "lambda:InvokeFunction"
            ],
            "Resource": [
                "arn:aws:lambda:us-east-1:012345678901:function:MapSampleProj-ReadFromSQSQueueLambda-1MY3M63RMJVA9",
                "arn:aws:lambda:us-east-1:012345678901:function:MapSampleProj-DeleteFromSQSQueueLambda-198J2839ZO5K2"
            ],
            "Effect": "Allow"
        },
        {
            "Action": [
                "dynamodb:PutItem"
            ],
            "Resource": [
                "arn:aws:dynamodb:us-east-1:012345678901:table/MapSampleProj-DDBTable-YJDJ1MKIN6C5"
            ],
            "Effect": "Allow"
        },
        {
            "Action": [
                "sns:Publish"
            ],
            "Resource": [
                "arn:aws:sns:us-east-1:012345678901:MapSampleProj-SNSTopic-1CQO4HQ3IR1KN"
            ],
            "Effect": "Allow"
        }
    ]
}

Informationen zur Konfiguration von IAM bei der Verwendung von Step Functions mit anderen AWS Diensten finden Sie unterIAM-Richtlinien für integrierte Dienste.