Memproses pesan volume tinggi dari Amazon SQS dengan alur kerja Step Functions Express

Proyek contoh ini menunjukkan cara menggunakan Alur Kerja AWS Step Functions Ekspres untuk memproses pesan atau data dari sumber peristiwa bervolume tinggi, seperti Amazon Simple Queue Service (Amazon). SQS Karena Alur Kerja Ekspres dapat dimulai dengan kecepatan yang sangat tinggi, Alur kerja tersebut ideal untuk pemrosesan peristiwa atau beban kerja streaming data volume tinggi.

Berikut adalah dua metode yang umum digunakan untuk menjalankan mesin status Anda dari sumber peristiwa:

Proyek sampel ini menggunakan metode kedua untuk memulai eksekusi setiap kali SQS antrian Amazon mengirim pesan. Anda dapat menggunakan konfigurasi serupa untuk memicu eksekusi Alur Kerja Express dari sumber peristiwa lain, seperti Amazon Simple Storage Service (Amazon S3), Amazon DynamoDB, dan Amazon Kinesis.

Untuk informasi selengkapnya tentang Alur Kerja Express dan integrasi layanan Step Functions, lihat di bawah ini:

Langkah 1: Buat mesin negara bagian dan sumber daya penyediaan

Langkah 2: Memicu eksekusi mesin negara

Contoh Kode Fungsi Lambda

Berikut ini adalah kode fungsi Lambda yang menunjukkan bagaimana fungsi Lambda memulai eksekusi mesin status menggunakan. 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
    )

Contoh Kode Mesin Status

Alur Kerja Express dalam proyek sampel ini terdiri dari satu set fungsi Lambda untuk pemrosesan teks.

Untuk informasi selengkapnya tentang AWS Step Functions cara mengontrol AWS layanan lain, lihatMengintegrasikan layanan lain dengan 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
    }
  }
}

IAMContoh

Kebijakan contoh AWS Identity and Access Management (IAM) yang dihasilkan oleh proyek sampel ini mencakup hak istimewa paling sedikit yang diperlukan untuk mengeksekusi mesin negara dan sumber daya terkait. Kami menyarankan Anda hanya menyertakan izin yang diperlukan dalam IAM kebijakan Anda.

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

Kebijakan berikut memastikan bahwa ada izin yang cukup untuk CloudWatch Log.

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

Untuk informasi tentang cara mengonfigurasi IAM saat menggunakan Step Functions dengan AWS layanan lain, lihatBagaimana Step Functions menghasilkan IAM kebijakan untuk layanan terintegrasi.