Prasyarat Buat peran eksekusi Buat fungsi Uji fungsi Lambda Buat tabel DynamoDB dengan aliran yang diaktifkan Menambahkan sumber acara di AWS Lambda Uji penyiapan Langkah selanjutnya Bersihkan sumber daya Anda

Tutorial: Menggunakan AWS Lambda dengan aliran Amazon DynamoDB

Dalam tutorial ini, Anda membuat fungsi Lambda untuk menggunakan kejadian dari aliran Amazon DynamoDB.

Prasyarat

Jika Anda belum menginstal AWS Command Line Interface, ikuti langkah-langkah di Meng instal atau memperbarui versi terbaru AWS CLI untuk menginstalnya.

Tutorial ini membutuhkan terminal baris perintah atau shell untuk menjalankan perintah. Di Linux dan macOS, gunakan shell dan manajer paket pilihan Anda.

catatan

Di Windows, beberapa perintah Bash CLI yang biasa Anda gunakan dengan Lambda (sepertizip) tidak didukung oleh terminal bawaan sistem operasi. Untuk mendapatkan Windows-integrated versi Ubuntu dan Bash, instal Subsistem Windows untuk Linux.

Buat peran eksekusi

Buat peran eksekusi yang memberikan izin fungsi Anda untuk mengakses AWS sumber daya.

Untuk membuat peran eksekusi

Buka halaman peran di konsol IAM.
Pilih Buat peran.
Buat peran dengan properti berikut.
- Entitas tepercaya – Lambda.
- Izin – AWSLambdaDynamoDBExecutionRole.
- Nama peran – lambda-dynamodb-role.

Memiliki AWSLambdaDynamoDBExecutionRole izin yang dibutuhkan fungsi untuk membaca item dari DynamoDB dan menulis log ke CloudWatch Log.

Buat fungsi

Buat fungsi Lambda yang memproses peristiwa DynamoDB Anda. Kode fungsi menulis beberapa data peristiwa masuk ke CloudWatch Log.

.NET

SDK untuk .NET

catatan

Ada lebih banyak tentang GitHub. Temukan contoh lengkapnya dan pelajari cara mengatur dan menjalankannya di repositori contoh Nirserver.

Mengkonsumsi peristiwa DynamoDB dengan Lambda menggunakan.NET.


// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
using System.Text.Json;
using System.Text;
using Amazon.Lambda.Core;
using Amazon.Lambda.DynamoDBEvents;

// Assembly attribute to enable the Lambda function's JSON input to be converted into a .NET class.
[assembly: LambdaSerializer(typeof(Amazon.Lambda.Serialization.SystemTextJson.DefaultLambdaJsonSerializer))]

namespace AWSLambda_DDB;

public class Function
{
    public void FunctionHandler(DynamoDBEvent dynamoEvent, ILambdaContext context)
    {
        context.Logger.LogInformation($"Beginning to process {dynamoEvent.Records.Count} records...");

        foreach (var record in dynamoEvent.Records)
        {
            context.Logger.LogInformation($"Event ID: {record.EventID}");
            context.Logger.LogInformation($"Event Name: {record.EventName}");

            context.Logger.LogInformation(JsonSerializer.Serialize(record));
        }

        context.Logger.LogInformation("Stream processing complete.");
    }
}

Go

SDK untuk Go V2

catatan

Ada lebih banyak tentang GitHub. Temukan contoh lengkapnya dan pelajari cara mengatur dan menjalankannya di repositori contoh Nirserver.

Mengkonsumsi peristiwa DynamoDB dengan Lambda menggunakan Go.


// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
package main

import (
	"context"
	"github.com/aws/aws-lambda-go/lambda"
	"github.com/aws/aws-lambda-go/events"
	"fmt"
)

func HandleRequest(ctx context.Context, event events.DynamoDBEvent) (*string, error) {
	if len(event.Records) == 0 {
		return nil, fmt.Errorf("received empty event")
	}

	for _, record := range event.Records {
	 	LogDynamoDBRecord(record)
	}

	message := fmt.Sprintf("Records processed: %d", len(event.Records))
	return &message, nil
}

func main() {
	lambda.Start(HandleRequest)
}

func LogDynamoDBRecord(record events.DynamoDBEventRecord){
	fmt.Println(record.EventID)
	fmt.Println(record.EventName)
	fmt.Printf("%+v\n", record.Change)
}

Java

SDK untuk Java 2.x

catatan

Ada lebih banyak tentang GitHub. Temukan contoh lengkapnya dan pelajari cara mengatur dan menjalankannya di repositori contoh Nirserver.

Mengkonsumsi peristiwa DynamoDB dengan Lambda menggunakan Java.


import com.amazonaws.services.lambda.runtime.Context;
import com.amazonaws.services.lambda.runtime.RequestHandler;
import com.amazonaws.services.lambda.runtime.events.DynamodbEvent;
import com.amazonaws.services.lambda.runtime.events.DynamodbEvent.DynamodbStreamRecord;
import com.google.gson.Gson;
import com.google.gson.GsonBuilder;

public class example implements RequestHandler<DynamodbEvent, Void> {

    private static final Gson GSON = new GsonBuilder().setPrettyPrinting().create();

    @Override
    public Void handleRequest(DynamodbEvent event, Context context) {
        System.out.println(GSON.toJson(event));
        event.getRecords().forEach(this::logDynamoDBRecord);
        return null;
    }

    private void logDynamoDBRecord(DynamodbStreamRecord record) {
        System.out.println(record.getEventID());
        System.out.println(record.getEventName());
        System.out.println("DynamoDB Record: " + GSON.toJson(record.getDynamodb()));
    }
}

JavaScript

SDK untuk JavaScript (v3)

catatan

Ada lebih banyak tentang GitHub. Temukan contoh lengkapnya dan pelajari cara mengatur dan menjalankannya di repositori contoh Nirserver.

Mengkonsumsi peristiwa DynamoDB dengan menggunakan Lambda. JavaScript


// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
exports.handler = async (event, context) => {
    console.log(JSON.stringify(event, null, 2));
    event.Records.forEach(record => {
        logDynamoDBRecord(record);
    });
};

const logDynamoDBRecord = (record) => {
    console.log(record.eventID);
    console.log(record.eventName);
    console.log(`DynamoDB Record: ${JSON.stringify(record.dynamodb)}`);
};

Mengkonsumsi peristiwa DynamoDB dengan menggunakan Lambda. TypeScript


export const handler = async (event, context) => {
    console.log(JSON.stringify(event, null, 2));
    event.Records.forEach(record => {
        logDynamoDBRecord(record);
    });
}
const logDynamoDBRecord = (record) => {
    console.log(record.eventID);
    console.log(record.eventName);
    console.log(`DynamoDB Record: ${JSON.stringify(record.dynamodb)}`);
};

PHP

SDK untuk PHP

catatan

Ada lebih banyak tentang GitHub. Temukan contoh lengkapnya dan pelajari cara mengatur dan menjalankannya di repositori contoh Nirserver.

Mengkonsumsi peristiwa DynamoDB dengan Lambda menggunakan PHP.


<?php

# using bref/bref and bref/logger for simplicity

use Bref\Context\Context;
use Bref\Event\DynamoDb\DynamoDbEvent;
use Bref\Event\DynamoDb\DynamoDbHandler;
use Bref\Logger\StderrLogger;

require __DIR__ . '/vendor/autoload.php';

class Handler extends DynamoDbHandler
{
    private StderrLogger $logger;

    public function __construct(StderrLogger $logger)
    {
        $this->logger = $logger;
    }

    /**
     * @throws JsonException
     * @throws \Bref\Event\InvalidLambdaEvent
     */
    public function handleDynamoDb(DynamoDbEvent $event, Context $context): void
    {
        $this->logger->info("Processing DynamoDb table items");
        $records = $event->getRecords();

        foreach ($records as $record) {
            $eventName = $record->getEventName();
            $keys = $record->getKeys();
            $old = $record->getOldImage();
            $new = $record->getNewImage();
            
            $this->logger->info("Event Name:".$eventName."\n");
            $this->logger->info("Keys:". json_encode($keys)."\n");
            $this->logger->info("Old Image:". json_encode($old)."\n");
            $this->logger->info("New Image:". json_encode($new));
            
            // TODO: Do interesting work based on the new data

            // Any exception thrown will be logged and the invocation will be marked as failed
        }

        $totalRecords = count($records);
        $this->logger->info("Successfully processed $totalRecords items");
    }
}

$logger = new StderrLogger();
return new Handler($logger);

Python

SDK untuk Python (Boto3)

catatan

Ada lebih banyak tentang GitHub. Temukan contoh lengkapnya dan pelajari cara mengatur dan menjalankannya di repositori contoh Nirserver.

Mengkonsumsi peristiwa DynamoDB dengan Lambda menggunakan Python.



import json

def lambda_handler(event, context):
    print(json.dumps(event, indent=2))

    for record in event['Records']:
        log_dynamodb_record(record)

def log_dynamodb_record(record):
    print(record['eventID'])
    print(record['eventName'])
    print(f"DynamoDB Record: {json.dumps(record['dynamodb'])}")

Ruby

SDK untuk Ruby

catatan

Ada lebih banyak tentang GitHub. Temukan contoh lengkapnya dan pelajari cara mengatur dan menjalankannya di repositori contoh Nirserver.

Mengkonsumsi peristiwa DynamoDB dengan Lambda menggunakan Ruby.



def lambda_handler(event:, context:)
    return 'received empty event' if event['Records'].empty?
  
    event['Records'].each do |record|
      log_dynamodb_record(record)
    end
  
    "Records processed: #{event['Records'].length}"
  end
  
  def log_dynamodb_record(record)
    puts record['eventID']
    puts record['eventName']
    puts "DynamoDB Record: #{JSON.generate(record['dynamodb'])}"
  end

Rust

SDK for Rust

catatan

Ada lebih banyak tentang GitHub. Temukan contoh lengkapnya dan pelajari cara mengatur dan menjalankannya di repositori contoh Nirserver.

Mengkonsumsi peristiwa DynamoDB dengan Lambda menggunakan Rust.



use lambda_runtime::{service_fn, tracing, Error, LambdaEvent};
use aws_lambda_events::{
    event::dynamodb::{Event, EventRecord},
   };


// Built with the following dependencies:
//lambda_runtime = "0.11.1"
//serde_json = "1.0"
//tokio = { version = "1", features = ["macros"] }
//tracing = { version = "0.1", features = ["log"] }
//tracing-subscriber = { version = "0.3", default-features = false, features = ["fmt"] }
//aws_lambda_events = "0.15.0"

async fn function_handler(event: LambdaEvent<Event>) ->Result<(), Error> {
    
    let records = &event.payload.records;
    tracing::info!("event payload: {:?}",records);
    if records.is_empty() {
        tracing::info!("No records found. Exiting.");
        return Ok(());
    }

    for record in records{
        log_dynamo_dbrecord(record);
    }

    tracing::info!("Dynamo db records processed");

    // Prepare the response
    Ok(())

}

fn log_dynamo_dbrecord(record: &EventRecord)-> Result<(), Error>{
    tracing::info!("EventId: {}", record.event_id);
    tracing::info!("EventName: {}", record.event_name);
    tracing::info!("DynamoDB Record: {:?}", record.change );
    Ok(())

}

#[tokio::main]
async fn main() -> Result<(), Error> {
    tracing_subscriber::fmt()
    .with_max_level(tracing::Level::INFO)
    .with_target(false)
    .without_time()
    .init();

    let func = service_fn(function_handler);
    lambda_runtime::run(func).await?;
    Ok(())
    
}

Untuk membuat fungsi

Salin kode sampel ke file dengan nama example.js.
Buat paket deployment.
```
zip function.zip example.js
```

Buat fungsi Lambda dengan perintah create-function.


aws lambda create-function --function-name ProcessDynamoDBRecords \
    --zip-file fileb://function.zip --handler example.handler --runtime nodejs24.x \
    --role arn:aws:iam::111122223333:role/lambda-dynamodb-role

Uji fungsi Lambda

Pada langkah ini, Anda memanggil fungsi Lambda Anda secara manual menggunakan perintah invoke AWS Lambda CLI dan contoh peristiwa DynamoDB berikut. Salin berikut ini ke dalam file bernamainput.txt.

contoh input.txt


{
   "Records":[
      {
         "eventID":"1",
         "eventName":"INSERT",
         "eventVersion":"1.0",
         "eventSource":"aws:dynamodb",
         "awsRegion":"us-east-1",
         "dynamodb":{
            "Keys":{
               "Id":{
                  "N":"101"
               }
            },
            "NewImage":{
               "Message":{
                  "S":"New item!"
               },
               "Id":{
                  "N":"101"
               }
            },
            "SequenceNumber":"111",
            "SizeBytes":26,
            "StreamViewType":"NEW_AND_OLD_IMAGES"
         },
         "eventSourceARN":"stream-ARN"
      },
      {
         "eventID":"2",
         "eventName":"MODIFY",
         "eventVersion":"1.0",
         "eventSource":"aws:dynamodb",
         "awsRegion":"us-east-1",
         "dynamodb":{
            "Keys":{
               "Id":{
                  "N":"101"
               }
            },
            "NewImage":{
               "Message":{
                  "S":"This item has changed"
               },
               "Id":{
                  "N":"101"
               }
            },
            "OldImage":{
               "Message":{
                  "S":"New item!"
               },
               "Id":{
                  "N":"101"
               }
            },
            "SequenceNumber":"222",
            "SizeBytes":59,
            "StreamViewType":"NEW_AND_OLD_IMAGES"
         },
         "eventSourceARN":"stream-ARN"
      },
      {
         "eventID":"3",
         "eventName":"REMOVE",
         "eventVersion":"1.0",
         "eventSource":"aws:dynamodb",
         "awsRegion":"us-east-1",
         "dynamodb":{
            "Keys":{
               "Id":{
                  "N":"101"
               }
            },
            "OldImage":{
               "Message":{
                  "S":"This item has changed"
               },
               "Id":{
                  "N":"101"
               }
            },
            "SequenceNumber":"333",
            "SizeBytes":38,
            "StreamViewType":"NEW_AND_OLD_IMAGES"
         },
         "eventSourceARN":"stream-ARN"
      }
   ]
}

Jalankan perintah invoke berikut.


aws lambda invoke --function-name ProcessDynamoDBRecords \
    --cli-binary-format raw-in-base64-out \
    --payload file://input.txt outputfile.txt

cli-binary-formatOpsi ini diperlukan jika Anda menggunakan AWS CLI versi 2. Untuk menjadikan ini pengaturan default, jalankanaws configure set cli-binary-format raw-in-base64-out. Untuk informasi selengkapnya, lihat opsi baris perintah global yang AWS CLI didukung di Panduan AWS Command Line Interface Pengguna untuk Versi 2.

Fungsi mengembalikan message string dalam badan respons.

Verifikasikan output dalam outputfile.txt file.

Buat tabel DynamoDB dengan aliran yang diaktifkan

Buat tabel Amazon DynamoDB dengan aliran yang diaktifkan.

Untuk membuat tabel DynamoDB

Buka Konsol DynamoDB.
Pilih Buat tabel.
Buat tabel dengan pengaturan berikut.
- Nama tabel – lambda-dynamodb-stream
- Kunci utama – id (string)
Pilih Buat.

Untuk mengaktifkan stream

Buka Konsol DynamoDB.
Pilih Tables.
Pilih tabel lambda-dynamodb-stream.
Di bawah Ekspor dan aliran, pilih detail aliran DynamoDB.
Pilih Aktifkan.
Untuk Jenis tampilan, pilih Atribut kunci saja.
Pilih Aktifkan streaming.

Tulis ARN stream. Anda memerlukan ini di langkah berikutnya ketika Anda mengaitkan aliran dengan fungsi Lambda Anda. Untuk informasi selengkapnya tentang mengaktifkan aliran, lihat Menangkap aktivitas tabel dengan DynamoDB Streams.

Menambahkan sumber acara di AWS Lambda

Buat pemetaan sumber peristiwa di AWS Lambda. Pemetaan sumber peristiwa ini mengaitkan aliran DynamoDB dengan fungsi Lambda Anda. Setelah Anda membuat pemetaan sumber peristiwa ini, mul AWS Lambda ailah melakukan polling aliran.

Jalankan perintah AWS CLI create-event-source-mapping berikut. Setelah perintah dijalankan, catat UUID. Anda akan memerlukan UUID ini untuk merujuk ke pemetaan sumber kejadian dalam perintah apa pun, misalnya, saat menghapus pemetaan sumber kejadian.


aws lambda create-event-source-mapping --function-name ProcessDynamoDBRecords \
    --batch-size 100 --starting-position LATEST --event-source DynamoDB-stream-arn

Ini membuat pemetaan di antara aliran DynamoDB yang ditentukan dan fungsi Lambda. Anda dapat mengaitkan aliran DynamoDB dengan beberapa fungsi Lambda, dan mengaitkan fungsi Lambda yang sama dengan beberapa aliran. Namun, fungsi Lambda berbagi throughput baca untuk aliran yang mereka bagikan.

Anda bisa mendapatkan daftar pemetaan sumber kejadian dengan menjalankan perintah berikut.


aws lambda list-event-source-mappings

Daftar tersebut mengembalikan semua pemetaan sumber kejadian yang Anda buat, dan antara lain menampilkan LastProcessingResult untuk setiap pemetaan. Bidang ini digunakan untuk memberikan pesan informatif jika terjadi masalah. Nilai seperti No records processed (menunjukkan bahwa AWS Lambda belum memulai polling atau tidak ada catatan dalam aliran) dan OK (menunjukkan AWS Lambda berhasil membaca catatan dari aliran dan memanggil fungsi Lambda Anda) menunjukkan bahwa tidak ada masalah. Jika ada masalah, Anda akan menerima pesan kesalahan.

Jika Anda memiliki banyak pemetaan sumber kejadian, gunakan parameter nama fungsi untuk mempersempit hasil.


aws lambda list-event-source-mappings --function-name ProcessDynamoDBRecords

Uji penyiapan

Uji pengalaman end-to-end. Saat Anda melakukan pembaruan tabel, DynamoDB menuliskan catatan peristiwa ke aliran. Saat AWS Lambda melakukan polling terhadap aliran, ini mendeteksi catatan baru dalam aliran dan memanggil fungsi Lambda Anda atas nama Anda dengan mengirimkan peristiwa ke fungsi tersebut.

Di konsol DynamoDB, tambahkan, perbarui, dan hapus item di tabel. DynamoDB menuliskan catatan tindakan ini ke aliran.
AWS Lambda mensurvei aliran dan ketika mendeteksi pembaruan pada aliran, ia memanggil fungsi Lambda Anda dengan meneruskan data acara yang ditemukannya di aliran.
Fungsi Anda berjalan dan membuat log di Amazon CloudWatch. Anda dapat memverifikasi log yang dilaporkan di CloudWatch konsol Amazon.

Langkah selanjutnya

Tutorial ini menunjukkan dasar-dasar pemrosesan peristiwa aliran DynamoDB dengan Lambda. Untuk beban kerja produksi, pertimbangkan untuk menerapkan logika respons batch parsional untuk menangani kegagalan catatan individual dengan lebih efisien. Utilitas prosesor batch dari Powertools for AWS Lambda tersedia di Python,, .NET TypeScript, dan Java dan memberikan solusi yang kuat untuk ini, secara otomatis menangani kompleksitas respons batch parsional dan mengurangi jumlah percobaan ulang untuk catatan yang berhasil diproses.

Bersihkan sumber daya Anda

Sekarang Anda dapat menghapus sumber daya yang Anda buat untuk tutorial ini, kecuali Anda ingin mempertahankannya. Dengan menghapus AWS sumber daya yang tidak lagi Anda gunakan, Anda mencegah biaya yang tidak perlu kepada Anda Akun AWS.

Untuk menghapus fungsi Lambda

Buka halaman Fungsi di konsol Lambda.
Pilih fungsi yang Anda buat.
Pilih Tindakan, Hapus.
Ketik confirm di bidang input teks dan pilih H apus.

Untuk menghapus peran eksekusi

Buka halaman Peran dari konsol IAM.
Pilih peran eksekusi yang Anda buat.
Pilih Hapus.
Masukkan nama peran di bidang input teks dan pilih H apus.

Untuk menghapus tabel DynamoDB

Buka halaman Tabel di konsol DynamoDB.
Pilih tabel yang Anda buat.
Pilih Hapus.
Masukkan delete di kotak teks.
Pilih Hapus tabel.

Awas Javascript dinonaktifkan atau tidak tersedia di browser Anda.

Untuk menggunakan Dokumentasi AWS, Javascript harus diaktifkan. Lihat halaman Bantuan browser Anda untuk petunjuk.

Konvensi Dokumen

Pemfilteran peristiwa

EC2