翻訳は機械翻訳により提供されています。提供された翻訳内容と英語版の間で齟齬、不一致または矛盾がある場合、英語版が優先します。

Step Functions Express ワークフローSQSを使用して Amazon からの大量のメッセージを処理

このサンプルプロジェクトでは、 の使用方法を示します。 AWS Step Functions Express ワークフローは、Amazon Simple Queue Service (Amazon ) などの大量のイベントソースからのメッセージまたはデータを処理しますSQS。Express ワークフローは非常に高いレートで開始できるため、大容量のイベント処理やストリーミングデータワークロードに最適です。

イベントソースからステートマシンを実行するために一般的に使用される 2 つの方法は次のとおりです。

このサンプルプロジェクトでは、Amazon SQSキューがメッセージを送信するたびに 2 番目のメソッドを使用して実行を開始します。同様の設定を使用して、Amazon Simple Storage Service (Amazon S3)、Amazon DynamoDB、Amazon Kinesis などの他のイベントソースから Express ワークフローの実行をトリガーできます。

Express ワークフローと Step Functions サービス統合の詳細については、以下を参照してください。

ステップ 1: ステートマシンを作成する

次のステップは、前の選択肢によって異なります。

ステップ 2: ステートマシンの実行をトリガーする

Lambda 関数のコードの例

以下は、Lambda 関数の開始によって を使用してステートマシンの実行を開始する方法を示す Lambda 関数コードです。 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
    )

ステートマシンのコード例

このサンプルプロジェクトの Express ワークフローは、テキスト処理用の Lambda 関数のセットで構成されています。

方法の詳細については、 AWS Step Functions は他の を制御できます AWS サービスについては、「」を参照してくださいサービスと 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
    }
  }
}

IAM 例

この例では AWS Identity and Access Management サンプルプロジェクトによって生成された (IAM) ポリシーには、ステートマシンおよび関連リソースの実行に必要な最小特権が含まれています。IAM ポリシーに必要なアクセス許可のみを含めることをお勧めします。

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

次のポリシーは、 CloudWatch ログに十分なアクセス許可があることを保証します。

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

他の で Step Functions を使用するIAMときに を設定する方法の詳細については、「」を参照してください。 AWS サービスについては、「」を参照してくださいStep Functions が統合サービスのIAMポリシーを生成する方法。