# Lambda 无服务器示例
以下代码示例显示如何将 Lambda 与 AWS SDK 一起使用。
**Topics**
+ [使用 Lambda 函数连接到 Amazon RDS 数据库](example_serverless_connect_RDS_Lambda_section.md)
+ [通过 Kinesis 触发器调用 Lambda 函数](example_serverless_Kinesis_Lambda_section.md)
+ [通过 DynamoDB 触发器调用 Lambda 函数](example_serverless_DynamoDB_Lambda_section.md)
+ [通过 Amazon DocumentDB 触发器调用 Lambda 函数](example_serverless_DocumentDB_Lambda_section.md)
+ [通过 Amazon MSK 触发器调用 Lambda 函数](example_serverless_MSK_Lambda_section.md)
+ [通过 Amazon S3 触发器调用 Lambda 函数](example_serverless_S3_Lambda_section.md)
+ [通过 Amazon SNS 触发器调用 Lambda 函数](example_serverless_SNS_Lambda_section.md)
+ [通过 Amazon SQS 触发器调用 Lambda 函数](example_serverless_SQS_Lambda_section.md)
+ [通过 Kinesis 触发器报告 Lambda 函数批处理项目失败](example_serverless_Kinesis_Lambda_batch_item_failures_section.md)
+ [通过 DynamoDB 触发器报告 Lambda 函数批处理项目失败](example_serverless_DynamoDB_Lambda_batch_item_failures_section.md)
+ [报告使用 Amazon SQS 触发器进行 Lambda 函数批处理项目失败](example_serverless_SQS_Lambda_batch_item_failures_section.md)
# 使用 Lambda 函数连接到 Amazon RDS 数据库
以下代码示例显示如何实现连接到 RDS 数据库的 Lambda 函数。该函数发出一个简单的数据库请求并返回结果。
------
#### [ .NET ]
**适用于 .NET 的 SDK**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/lambda-function-connect-rds-iam)存储库中查找完整示例，并了解如何进行设置和运行。
在 Lambda 函数中使用 .NET 连接到 Amazon RDS 数据库。
```
using System.Data;
using System.Text.Json;
using Amazon.Lambda.APIGatewayEvents;
using Amazon.Lambda.Core;
using MySql.Data.MySqlClient;
// 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 aws_rds;
public class InputModel
{
public string key1 { get; set; }
public string key2 { get; set; }
}
public class Function
{
///
// Handles the Lambda function execution for connecting to RDS using IAM authentication.
///
/// The input event data passed to the Lambda function
/// The Lambda execution context that provides runtime information
/// A response object containing the execution result
public async Task FunctionHandler(APIGatewayProxyRequest request, ILambdaContext context)
{
// Sample Input: {"body": "{\"key1\":\"20\", \"key2\":\"25\"}"}
var input = JsonSerializer.Deserialize(request.Body);
/// Obtain authentication token
var authToken = RDSAuthTokenGenerator.GenerateAuthToken(
Environment.GetEnvironmentVariable("RDS_ENDPOINT"),
Convert.ToInt32(Environment.GetEnvironmentVariable("RDS_PORT")),
Environment.GetEnvironmentVariable("RDS_USERNAME")
);
/// Build the Connection String with the Token
string connectionString = $"Server={Environment.GetEnvironmentVariable("RDS_ENDPOINT")};" +
$"Port={Environment.GetEnvironmentVariable("RDS_PORT")};" +
$"Uid={Environment.GetEnvironmentVariable("RDS_USERNAME")};" +
$"Pwd={authToken};";
try
{
await using var connection = new MySqlConnection(connectionString);
await connection.OpenAsync();
const string sql = "SELECT @param1 + @param2 AS Sum";
await using var command = new MySqlCommand(sql, connection);
command.Parameters.AddWithValue("@param1", int.Parse(input.key1 ?? "0"));
command.Parameters.AddWithValue("@param2", int.Parse(input.key2 ?? "0"));
await using var reader = await command.ExecuteReaderAsync();
if (await reader.ReadAsync())
{
int result = reader.GetInt32("Sum");
//Sample Response: {"statusCode":200,"body":"{\"message\":\"The sum is: 45\"}","isBase64Encoded":false}
return new APIGatewayProxyResponse
{
StatusCode = 200,
Body = JsonSerializer.Serialize(new { message = $"The sum is: {result}" })
};
}
}
catch (Exception ex)
{
Console.WriteLine($"Error: {ex.Message}");
}
return new APIGatewayProxyResponse
{
StatusCode = 500,
Body = JsonSerializer.Serialize(new { error = "Internal server error" })
};
}
}
```
------
#### [ Go ]
**适用于 Go 的 SDK V2**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/lambda-function-connect-rds-iam)存储库中查找完整示例，并了解如何进行设置和运行。
在 Lambda 函数中使用 Go 连接到 Amazon RDS 数据库。
```
/*
Golang v2 code here.
*/
package main
import (
"context"
"database/sql"
"encoding/json"
"fmt"
"os"
"github.com/aws/aws-lambda-go/lambda"
"github.com/aws/aws-sdk-go-v2/config"
"github.com/aws/aws-sdk-go-v2/feature/rds/auth"
_ "github.com/go-sql-driver/mysql"
)
type MyEvent struct {
Name string `json:"name"`
}
func HandleRequest(event *MyEvent) (map[string]interface{}, error) {
var dbName string = os.Getenv("DatabaseName")
var dbUser string = os.Getenv("DatabaseUser")
var dbHost string = os.Getenv("DBHost") // Add hostname without https
var dbPort int = os.Getenv("Port") // Add port number
var dbEndpoint string = fmt.Sprintf("%s:%d", dbHost, dbPort)
var region string = os.Getenv("AWS_REGION")
cfg, err := config.LoadDefaultConfig(context.TODO())
if err != nil {
panic("configuration error: " + err.Error())
}
authenticationToken, err := auth.BuildAuthToken(
context.TODO(), dbEndpoint, region, dbUser, cfg.Credentials)
if err != nil {
panic("failed to create authentication token: " + err.Error())
}
dsn := fmt.Sprintf("%s:%s@tcp(%s)/%s?tls=true&allowCleartextPasswords=true",
dbUser, authenticationToken, dbEndpoint, dbName,
)
db, err := sql.Open("mysql", dsn)
if err != nil {
panic(err)
}
defer db.Close()
var sum int
err = db.QueryRow("SELECT ?+? AS sum", 3, 2).Scan(&sum)
if err != nil {
panic(err)
}
s := fmt.Sprint(sum)
message := fmt.Sprintf("The selected sum is: %s", s)
messageBytes, err := json.Marshal(message)
if err != nil {
return nil, err
}
messageString := string(messageBytes)
return map[string]interface{}{
"statusCode": 200,
"headers": map[string]string{"Content-Type": "application/json"},
"body": messageString,
}, nil
}
func main() {
lambda.Start(HandleRequest)
}
```
------
#### [ Java ]
**适用于 Java 的 SDK 2.x**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/lambda-function-connect-rds-iam)存储库中查找完整示例，并了解如何进行设置和运行。
在 Lambda 函数中使用 Java 连接到 Amazon RDS 数据库。
```
import com.amazonaws.services.lambda.runtime.Context;
import com.amazonaws.services.lambda.runtime.RequestHandler;
import com.amazonaws.services.lambda.runtime.events.APIGatewayProxyRequestEvent;
import com.amazonaws.services.lambda.runtime.events.APIGatewayProxyResponseEvent;
import software.amazon.awssdk.auth.credentials.DefaultCredentialsProvider;
import software.amazon.awssdk.regions.Region;
import software.amazon.awssdk.services.rdsdata.RdsDataClient;
import software.amazon.awssdk.services.rdsdata.model.ExecuteStatementRequest;
import software.amazon.awssdk.services.rdsdata.model.ExecuteStatementResponse;
import software.amazon.awssdk.services.rdsdata.model.Field;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
public class RdsLambdaHandler implements RequestHandler {
@Override
public APIGatewayProxyResponseEvent handleRequest(APIGatewayProxyRequestEvent event, Context context) {
APIGatewayProxyResponseEvent response = new APIGatewayProxyResponseEvent();
try {
// Obtain auth token
String token = createAuthToken();
// Define connection configuration
String connectionString = String.format("jdbc:mysql://%s:%s/%s?useSSL=true&requireSSL=true",
System.getenv("ProxyHostName"),
System.getenv("Port"),
System.getenv("DBName"));
// Establish a connection to the database
try (Connection connection = DriverManager.getConnection(connectionString, System.getenv("DBUserName"), token);
PreparedStatement statement = connection.prepareStatement("SELECT ? + ? AS sum")) {
statement.setInt(1, 3);
statement.setInt(2, 2);
try (ResultSet resultSet = statement.executeQuery()) {
if (resultSet.next()) {
int sum = resultSet.getInt("sum");
response.setStatusCode(200);
response.setBody("The selected sum is: " + sum);
}
}
}
} catch (Exception e) {
response.setStatusCode(500);
response.setBody("Error: " + e.getMessage());
}
return response;
}
private String createAuthToken() {
// Create RDS Data Service client
RdsDataClient rdsDataClient = RdsDataClient.builder()
.region(Region.of(System.getenv("AWS_REGION")))
.credentialsProvider(DefaultCredentialsProvider.create())
.build();
// Define authentication request
ExecuteStatementRequest request = ExecuteStatementRequest.builder()
.resourceArn(System.getenv("ProxyHostName"))
.secretArn(System.getenv("DBUserName"))
.database(System.getenv("DBName"))
.sql("SELECT 'RDS IAM Authentication'")
.build();
// Execute request and obtain authentication token
ExecuteStatementResponse response = rdsDataClient.executeStatement(request);
Field tokenField = response.records().get(0).get(0);
return tokenField.stringValue();
}
}
```
------
#### [ JavaScript ]
**SDK for JavaScript（v3）**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/lambda-function-connect-rds-iam)存储库中查找完整示例，并了解如何进行设置和运行。
在 Lambda 函数中使用 JavaScript 连接到 Amazon RDS 数据库。
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
/*
Node.js code here.
*/
// ES6+ example
import { Signer } from "@aws-sdk/rds-signer";
import mysql from 'mysql2/promise';
async function createAuthToken() {
// Define connection authentication parameters
const dbinfo = {
hostname: process.env.ProxyHostName,
port: process.env.Port,
username: process.env.DBUserName,
region: process.env.AWS_REGION,
}
// Create RDS Signer object
const signer = new Signer(dbinfo);
// Request authorization token from RDS, specifying the username
const token = await signer.getAuthToken();
return token;
}
async function dbOps() {
// Obtain auth token
const token = await createAuthToken();
// Define connection configuration
let connectionConfig = {
host: process.env.ProxyHostName,
user: process.env.DBUserName,
password: token,
database: process.env.DBName,
ssl: 'Amazon RDS'
}
// Create the connection to the DB
const conn = await mysql.createConnection(connectionConfig);
// Obtain the result of the query
const [res,] = await conn.execute('select ?+? as sum', [3, 2]);
return res;
}
export const handler = async (event) => {
// Execute database flow
const result = await dbOps();
// Return result
return {
statusCode: 200,
body: JSON.stringify("The selected sum is: " + result[0].sum)
}
};
```
在 Lambda 函数中使用 TypeScript 连接到 Amazon RDS 数据库。
```
import { Signer } from "@aws-sdk/rds-signer";
import mysql from 'mysql2/promise';
// RDS settings
// Using '!' (non-null assertion operator) to tell the TypeScript compiler that the DB settings are not null or undefined,
const proxy_host_name = process.env.PROXY_HOST_NAME!
const port = parseInt(process.env.PORT!)
const db_name = process.env.DB_NAME!
const db_user_name = process.env.DB_USER_NAME!
const aws_region = process.env.AWS_REGION!
async function createAuthToken(): Promise {
// Create RDS Signer object
const signer = new Signer({
hostname: proxy_host_name,
port: port,
region: aws_region,
username: db_user_name
});
// Request authorization token from RDS, specifying the username
const token = await signer.getAuthToken();
return token;
}
async function dbOps(): Promise {
try {
// Obtain auth token
const token = await createAuthToken();
const conn = await mysql.createConnection({
host: proxy_host_name,
user: db_user_name,
password: token,
database: db_name,
ssl: 'Amazon RDS' // Ensure you have the CA bundle for SSL connection
});
const [rows, fields] = await conn.execute('SELECT ? + ? AS sum', [3, 2]);
console.log('result:', rows);
return rows;
}
catch (err) {
console.log(err);
}
}
export const lambdaHandler = async (event: any): Promise<{ statusCode: number; body: string }> => {
// Execute database flow
const result = await dbOps();
// Return error is result is undefined
if (result == undefined)
return {
statusCode: 500,
body: JSON.stringify(`Error with connection to DB host`)
}
// Return result
return {
statusCode: 200,
body: JSON.stringify(`The selected sum is: ${result[0].sum}`)
};
};
```
------
#### [ PHP ]
**适用于 PHP 的 SDK**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/lambda-function-connect-rds-iam)存储库中查找完整示例，并了解如何进行设置和运行。
在 Lambda 函数中使用 PHP 连接到 Amazon RDS 数据库。
```
logger = $logger;
}
private function getAuthToken(): string {
// Define connection authentication parameters
$dbConnection = [
'hostname' => getenv('DB_HOSTNAME'),
'port' => getenv('DB_PORT'),
'username' => getenv('DB_USERNAME'),
'region' => getenv('AWS_REGION'),
];
// Create RDS AuthTokenGenerator object
$generator = new AuthTokenGenerator(CredentialProvider::defaultProvider());
// Request authorization token from RDS, specifying the username
return $generator->createToken(
$dbConnection['hostname'] . ':' . $dbConnection['port'],
$dbConnection['region'],
$dbConnection['username']
);
}
private function getQueryResults() {
// Obtain auth token
$token = $this->getAuthToken();
// Define connection configuration
$connectionConfig = [
'host' => getenv('DB_HOSTNAME'),
'user' => getenv('DB_USERNAME'),
'password' => $token,
'database' => getenv('DB_NAME'),
];
// Create the connection to the DB
$conn = new PDO(
"mysql:host={$connectionConfig['host']};dbname={$connectionConfig['database']}",
$connectionConfig['user'],
$connectionConfig['password'],
[
PDO::MYSQL_ATTR_SSL_CA => '/path/to/rds-ca-2019-root.pem',
PDO::MYSQL_ATTR_SSL_VERIFY_SERVER_CERT => true,
]
);
// Obtain the result of the query
$stmt = $conn->prepare('SELECT ?+? AS sum');
$stmt->execute([3, 2]);
return $stmt->fetch(PDO::FETCH_ASSOC);
}
/**
* @param mixed $event
* @param Context $context
* @return array
*/
public function handle(mixed $event, Context $context): array
{
$this->logger->info("Processing query");
// Execute database flow
$result = $this->getQueryResults();
return [
'sum' => $result['sum']
];
}
}
$logger = new StderrLogger();
return new Handler($logger);
```
------
#### [ Python ]
**适用于 Python 的 SDK（Boto3）**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/lambda-function-connect-rds-iam)存储库中查找完整示例，并了解如何进行设置和运行。
在 Lambda 函数中使用 Python 连接到 Amazon RDS 数据库。
```
import json
import os
import boto3
import pymysql
# RDS settings
proxy_host_name = os.environ['PROXY_HOST_NAME']
port = int(os.environ['PORT'])
db_name = os.environ['DB_NAME']
db_user_name = os.environ['DB_USER_NAME']
aws_region = os.environ['AWS_REGION']
# Fetch RDS Auth Token
def get_auth_token():
client = boto3.client('rds')
token = client.generate_db_auth_token(
DBHostname=proxy_host_name,
Port=port
DBUsername=db_user_name
Region=aws_region
)
return token
def lambda_handler(event, context):
token = get_auth_token()
try:
connection = pymysql.connect(
host=proxy_host_name,
user=db_user_name,
password=token,
db=db_name,
port=port,
ssl={'ca': 'Amazon RDS'} # Ensure you have the CA bundle for SSL connection
)
with connection.cursor() as cursor:
cursor.execute('SELECT %s + %s AS sum', (3, 2))
result = cursor.fetchone()
return result
except Exception as e:
return (f"Error: {str(e)}") # Return an error message if an exception occurs
```
------
#### [ Ruby ]
**适用于 Ruby 的 SDK**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/lambda-function-connect-rds-iam)存储库中查找完整示例，并了解如何进行设置和运行。
在 Lambda 函数中使用 Ruby 连接到 Amazon RDS 数据库。
```
# Ruby code here.
require 'aws-sdk-rds'
require 'json'
require 'mysql2'
def lambda_handler(event:, context:)
endpoint = ENV['DBEndpoint'] # Add the endpoint without https"
port = ENV['Port'] # 3306
user = ENV['DBUser']
region = ENV['DBRegion'] # 'us-east-1'
db_name = ENV['DBName']
credentials = Aws::Credentials.new(
ENV['AWS_ACCESS_KEY_ID'],
ENV['AWS_SECRET_ACCESS_KEY'],
ENV['AWS_SESSION_TOKEN']
)
rds_client = Aws::RDS::AuthTokenGenerator.new(
region: region,
credentials: credentials
)
token = rds_client.auth_token(
endpoint: endpoint+ ':' + port,
user_name: user,
region: region
)
begin
conn = Mysql2::Client.new(
host: endpoint,
username: user,
password: token,
port: port,
database: db_name,
sslca: '/var/task/global-bundle.pem',
sslverify: true,
enable_cleartext_plugin: true
)
a = 3
b = 2
result = conn.query("SELECT #{a} + #{b} AS sum").first['sum']
puts result
conn.close
{
statusCode: 200,
body: result.to_json
}
rescue => e
puts "Database connection failed due to #{e}"
end
end
```
------
#### [ Rust ]
**适用于 Rust 的 SDK**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/lambda-function-connect-rds-iam)存储库中查找完整示例，并了解如何进行设置和运行。
在 Lambda 函数中使用 Rust 连接到 Amazon RDS 数据库。
```
use aws_config::BehaviorVersion;
use aws_credential_types::provider::ProvideCredentials;
use aws_sigv4::{
http_request::{sign, SignableBody, SignableRequest, SigningSettings},
sign::v4,
};
use lambda_runtime::{run, service_fn, Error, LambdaEvent};
use serde_json::{json, Value};
use sqlx::postgres::PgConnectOptions;
use std::env;
use std::time::{Duration, SystemTime};
const RDS_CERTS: &[u8] = include_bytes!("global-bundle.pem");
async fn generate_rds_iam_token(
db_hostname: &str,
port: u16,
db_username: &str,
) -> Result {
let config = aws_config::load_defaults(BehaviorVersion::v2024_03_28()).await;
let credentials = config
.credentials_provider()
.expect("no credentials provider found")
.provide_credentials()
.await
.expect("unable to load credentials");
let identity = credentials.into();
let region = config.region().unwrap().to_string();
let mut signing_settings = SigningSettings::default();
signing_settings.expires_in = Some(Duration::from_secs(900));
signing_settings.signature_location = aws_sigv4::http_request::SignatureLocation::QueryParams;
let signing_params = v4::SigningParams::builder()
.identity(&identity)
.region(®ion)
.name("rds-db")
.time(SystemTime::now())
.settings(signing_settings)
.build()?;
let url = format!(
"https://{db_hostname}:{port}/?Action=connect&DBUser={db_user}",
db_hostname = db_hostname,
port = port,
db_user = db_username
);
let signable_request =
SignableRequest::new("GET", &url, std::iter::empty(), SignableBody::Bytes(&[]))
.expect("signable request");
let (signing_instructions, _signature) =
sign(signable_request, &signing_params.into())?.into_parts();
let mut url = url::Url::parse(&url).unwrap();
for (name, value) in signing_instructions.params() {
url.query_pairs_mut().append_pair(name, &value);
}
let response = url.to_string().split_off("https://".len());
Ok(response)
}
#[tokio::main]
async fn main() -> Result<(), Error> {
run(service_fn(handler)).await
}
async fn handler(_event: LambdaEvent) -> Result {
let db_host = env::var("DB_HOSTNAME").expect("DB_HOSTNAME must be set");
let db_port = env::var("DB_PORT")
.expect("DB_PORT must be set")
.parse::()
.expect("PORT must be a valid number");
let db_name = env::var("DB_NAME").expect("DB_NAME must be set");
let db_user_name = env::var("DB_USERNAME").expect("DB_USERNAME must be set");
let token = generate_rds_iam_token(&db_host, db_port, &db_user_name).await?;
let opts = PgConnectOptions::new()
.host(&db_host)
.port(db_port)
.username(&db_user_name)
.password(&token)
.database(&db_name)
.ssl_root_cert_from_pem(RDS_CERTS.to_vec())
.ssl_mode(sqlx::postgres::PgSslMode::Require);
let pool = sqlx::postgres::PgPoolOptions::new()
.connect_with(opts)
.await?;
let result: i32 = sqlx::query_scalar("SELECT $1 + $2")
.bind(3)
.bind(2)
.fetch_one(&pool)
.await?;
println!("Result: {:?}", result);
Ok(json!({
"statusCode": 200,
"content-type": "text/plain",
"body": format!("The selected sum is: {result}")
}))
}
```
------
有关 AWS SDK 开发人员指南和代码示例的完整列表，请参阅 [将 Lambda 与 AWS SDK 配合使用](sdk-general-information-section.md) 本主题还包括有关入门的信息以及有关先前的 SDK 版本的详细信息。
# 通过 Kinesis 触发器调用 Lambda 函数
以下代码示例演示了如何实现一个 Lambda 函数，该函数接收通过接收来自 Kinesis 流的记录而触发的事件。该函数检索 Kinesis 有效负载，将 Base64 解码，并记录下记录内容。
------
#### [ .NET ]
**适用于 .NET 的 SDK**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-kinesis-to-lambda)存储库中查找完整示例，并了解如何进行设置和运行。
通过 .NET 将 Kinesis 事件与 Lambda 结合使用。
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
using System.Text;
using Amazon.Lambda.Core;
using Amazon.Lambda.KinesisEvents;
using AWS.Lambda.Powertools.Logging;
// 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 KinesisIntegrationSampleCode;
public class Function
{
// Powertools Logger requires an environment variables against your function
// POWERTOOLS_SERVICE_NAME
[Logging(LogEvent = true)]
public async Task FunctionHandler(KinesisEvent evnt, ILambdaContext context)
{
if (evnt.Records.Count == 0)
{
Logger.LogInformation("Empty Kinesis Event received");
return;
}
foreach (var record in evnt.Records)
{
try
{
Logger.LogInformation($"Processed Event with EventId: {record.EventId}");
string data = await GetRecordDataAsync(record.Kinesis, context);
Logger.LogInformation($"Data: {data}");
// TODO: Do interesting work based on the new data
}
catch (Exception ex)
{
Logger.LogError($"An error occurred {ex.Message}");
throw;
}
}
Logger.LogInformation($"Successfully processed {evnt.Records.Count} records.");
}
private async Task GetRecordDataAsync(KinesisEvent.Record record, ILambdaContext context)
{
byte[] bytes = record.Data.ToArray();
string data = Encoding.UTF8.GetString(bytes);
await Task.CompletedTask; //Placeholder for actual async work
return data;
}
}
```
------
#### [ Go ]
**适用于 Go 的 SDK V2**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-kinesis-to-lambda)存储库中查找完整示例，并了解如何进行设置和运行。
使用 Go 将 Kinesis 事件与 Lambda 结合使用。
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
package main
import (
"context"
"log"
"github.com/aws/aws-lambda-go/events"
"github.com/aws/aws-lambda-go/lambda"
)
func handler(ctx context.Context, kinesisEvent events.KinesisEvent) error {
if len(kinesisEvent.Records) == 0 {
log.Printf("empty Kinesis event received")
return nil
}
for _, record := range kinesisEvent.Records {
log.Printf("processed Kinesis event with EventId: %v", record.EventID)
recordDataBytes := record.Kinesis.Data
recordDataText := string(recordDataBytes)
log.Printf("record data: %v", recordDataText)
// TODO: Do interesting work based on the new data
}
log.Printf("successfully processed %v records", len(kinesisEvent.Records))
return nil
}
func main() {
lambda.Start(handler)
}
```
------
#### [ Java ]
**适用于 Java 的 SDK 2.x**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-kinesis-to-lambda)存储库中查找完整示例，并了解如何进行设置和运行。
使用 Java 将 Kinesis 事件与 Lambda 结合使用。
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
package example;
import com.amazonaws.services.lambda.runtime.Context;
import com.amazonaws.services.lambda.runtime.LambdaLogger;
import com.amazonaws.services.lambda.runtime.RequestHandler;
import com.amazonaws.services.lambda.runtime.events.KinesisEvent;
public class Handler implements RequestHandler {
@Override
public Void handleRequest(final KinesisEvent event, final Context context) {
LambdaLogger logger = context.getLogger();
if (event.getRecords().isEmpty()) {
logger.log("Empty Kinesis Event received");
return null;
}
for (KinesisEvent.KinesisEventRecord record : event.getRecords()) {
try {
logger.log("Processed Event with EventId: "+record.getEventID());
String data = new String(record.getKinesis().getData().array());
logger.log("Data:"+ data);
// TODO: Do interesting work based on the new data
}
catch (Exception ex) {
logger.log("An error occurred:"+ex.getMessage());
throw ex;
}
}
logger.log("Successfully processed:"+event.getRecords().size()+" records");
return null;
}
}
```
------
#### [ JavaScript ]
**SDK for JavaScript（v3）**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/blob/main/integration-kinesis-to-lambda)存储库中查找完整示例，并了解如何进行设置和运行。
通过 JavaScript 将 Kinesis 事件与 Lambda 结合使用。
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
exports.handler = async (event, context) => {
for (const record of event.Records) {
try {
console.log(`Processed Kinesis Event - EventID: ${record.eventID}`);
const recordData = await getRecordDataAsync(record.kinesis);
console.log(`Record Data: ${recordData}`);
// TODO: Do interesting work based on the new data
} catch (err) {
console.error(`An error occurred ${err}`);
throw err;
}
}
console.log(`Successfully processed ${event.Records.length} records.`);
};
async function getRecordDataAsync(payload) {
var data = Buffer.from(payload.data, "base64").toString("utf-8");
await Promise.resolve(1); //Placeholder for actual async work
return data;
}
```
通过 TypeScript 将 Kinesis 事件与 Lambda 结合使用。
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
import {
KinesisStreamEvent,
Context,
KinesisStreamHandler,
KinesisStreamRecordPayload,
} from "aws-lambda";
import { Buffer } from "buffer";
import { Logger } from "@aws-lambda-powertools/logger";
const logger = new Logger({
logLevel: "INFO",
serviceName: "kinesis-stream-handler-sample",
});
export const functionHandler: KinesisStreamHandler = async (
event: KinesisStreamEvent,
context: Context
): Promise => {
for (const record of event.Records) {
try {
logger.info(`Processed Kinesis Event - EventID: ${record.eventID}`);
const recordData = await getRecordDataAsync(record.kinesis);
logger.info(`Record Data: ${recordData}`);
// TODO: Do interesting work based on the new data
} catch (err) {
logger.error(`An error occurred ${err}`);
throw err;
}
logger.info(`Successfully processed ${event.Records.length} records.`);
}
};
async function getRecordDataAsync(
payload: KinesisStreamRecordPayload
): Promise {
var data = Buffer.from(payload.data, "base64").toString("utf-8");
await Promise.resolve(1); //Placeholder for actual async work
return data;
}
```
------
#### [ PHP ]
**适用于 PHP 的 SDK**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-kinesis-to-lambda)存储库中查找完整示例，并了解如何进行设置和运行。
通过 PHP 将 Kinesis 事件与 Lambda 结合使用。
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
logger = $logger;
}
/**
* @throws JsonException
* @throws \Bref\Event\InvalidLambdaEvent
*/
public function handleKinesis(KinesisEvent $event, Context $context): void
{
$this->logger->info("Processing records");
$records = $event->getRecords();
foreach ($records as $record) {
$data = $record->getData();
$this->logger->info(json_encode($data));
// 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 records");
}
}
$logger = new StderrLogger();
return new Handler($logger);
```
------
#### [ Python ]
**适用于 Python 的 SDK（Boto3）**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-kinesis-to-lambda)存储库中查找完整示例，并了解如何进行设置和运行。
使用 Python 将 Kinesis 事件与 Lambda 结合使用。
```
# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
# SPDX-License-Identifier: Apache-2.0
import base64
def lambda_handler(event, context):
for record in event['Records']:
try:
print(f"Processed Kinesis Event - EventID: {record['eventID']}")
record_data = base64.b64decode(record['kinesis']['data']).decode('utf-8')
print(f"Record Data: {record_data}")
# TODO: Do interesting work based on the new data
except Exception as e:
print(f"An error occurred {e}")
raise e
print(f"Successfully processed {len(event['Records'])} records.")
```
------
#### [ Ruby ]
**适用于 Ruby 的 SDK**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-kinesis-to-lambda)存储库中查找完整示例，并了解如何进行设置和运行。
通过 Ruby 将 Kinesis 事件与 Lambda 结合使用。
```
# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
# SPDX-License-Identifier: Apache-2.0
require 'aws-sdk'
def lambda_handler(event:, context:)
event['Records'].each do |record|
begin
puts "Processed Kinesis Event - EventID: #{record['eventID']}"
record_data = get_record_data_async(record['kinesis'])
puts "Record Data: #{record_data}"
# TODO: Do interesting work based on the new data
rescue => err
$stderr.puts "An error occurred #{err}"
raise err
end
end
puts "Successfully processed #{event['Records'].length} records."
end
def get_record_data_async(payload)
data = Base64.decode64(payload['data']).force_encoding('UTF-8')
# Placeholder for actual async work
# You can use Ruby's asynchronous programming tools like async/await or fibers here.
return data
end
```
------
#### [ Rust ]
**适用于 Rust 的 SDK**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-kinesis-to-lambda)存储库中查找完整示例，并了解如何进行设置和运行。
通过 Rust 将 Kinesis 事件与 Lambda 结合使用。
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
use aws_lambda_events::event::kinesis::KinesisEvent;
use lambda_runtime::{run, service_fn, Error, LambdaEvent};
async fn function_handler(event: LambdaEvent) -> Result<(), Error> {
if event.payload.records.is_empty() {
tracing::info!("No records found. Exiting.");
return Ok(());
}
event.payload.records.iter().for_each(|record| {
tracing::info!("EventId: {}",record.event_id.as_deref().unwrap_or_default());
let record_data = std::str::from_utf8(&record.kinesis.data);
match record_data {
Ok(data) => {
// log the record data
tracing::info!("Data: {}", data);
}
Err(e) => {
tracing::error!("Error: {}", e);
}
}
});
tracing::info!(
"Successfully processed {} records",
event.payload.records.len()
);
Ok(())
}
#[tokio::main]
async fn main() -> Result<(), Error> {
tracing_subscriber::fmt()
.with_max_level(tracing::Level::INFO)
// disable printing the name of the module in every log line.
.with_target(false)
// disabling time is handy because CloudWatch will add the ingestion time.
.without_time()
.init();
run(service_fn(function_handler)).await
}
```
------
有关 AWS SDK 开发人员指南和代码示例的完整列表，请参阅 [将 Lambda 与 AWS SDK 配合使用](sdk-general-information-section.md) 本主题还包括有关入门的信息以及有关先前的 SDK 版本的详细信息。
# 通过 DynamoDB 触发器调用 Lambda 函数
以下代码示例演示了如何实现一个 Lambda 函数，该函数接收通过接收来自 DynamoDB 流的记录而触发的事件。该函数检索 DynamoDB 有效负载，并记录下记录内容。
------
#### [ .NET ]
**适用于 .NET 的 SDK**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-ddb-to-lambda)存储库中查找完整示例，并了解如何进行设置和运行。
通过 .NET 将 DynamoDB 事件与 Lambda 结合使用。
```
// 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 ]
**适用于 Go 的 SDK V2**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-ddb-to-lambda)存储库中查找完整示例，并了解如何进行设置和运行。
使用 Go 将 DynamoDB 事件与 Lambda 结合使用。
```
// 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 ]
**适用于 Java 的 SDK 2.x**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-ddb-to-lambda)存储库中查找完整示例，并了解如何进行设置和运行。
使用 Java 将 DynamoDB 事件与 Lambda 结合使用。
```
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 {
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 for JavaScript（v3）**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-ddb-to-lambda)存储库中查找完整示例，并了解如何进行设置和运行。
使用 JavaScript 将 DynamoDB 事件与 Lambda 结合使用。
```
// 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)}`);
};
```
使用 TypeScript 将 DynamoDB 事件与 Lambda 结合使用。
```
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 ]
**适用于 PHP 的 SDK**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-ddb-to-lambda)存储库中查找完整示例，并了解如何进行设置和运行。
使用 PHP 将 DynamoDB 事件与 Lambda 结合使用。
```
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 ]
**适用于 Python 的 SDK（Boto3）**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-ddb-to-lambda)存储库中查找完整示例，并了解如何进行设置和运行。
通过 Python 将 DynamoDB 事件与 Lambda 结合使用。
```
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 ]
**适用于 Ruby 的 SDK**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-ddb-to-lambda)存储库中查找完整示例，并了解如何进行设置和运行。
通过 Ruby 将 DynamoDB 事件与 Lambda 结合使用。
```
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 ]
**适用于 Rust 的 SDK**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-ddb-to-lambda)存储库中查找完整示例，并了解如何进行设置和运行。
使用 Rust 将 DynamoDB 事件与 Lambda 结合使用。
```
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) ->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(())
}
```
------
有关 AWS SDK 开发人员指南和代码示例的完整列表，请参阅 [将 Lambda 与 AWS SDK 配合使用](sdk-general-information-section.md) 本主题还包括有关入门的信息以及有关先前的 SDK 版本的详细信息。
# 通过 Amazon DocumentDB 触发器调用 Lambda 函数
以下代码示例演示了如何实现一个 Lambda 函数，该函数接收通过接收来自 DynamoDB 更改流的记录而触发的事件。该函数检索 DocumentDB 有效负载，并记录下记录内容。
------
#### [ .NET ]
**适用于 .NET 的 SDK**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-docdb-to-lambda)存储库中查找完整示例，并了解如何进行设置和运行。
使用 .NET 将 Amazon DocumentDB 事件与 Lambda 结合使用。
```
using Amazon.Lambda.Core;
using System.Text.Json;
using System;
using System.Collections.Generic;
using System.Text.Json.Serialization;
//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 LambdaDocDb;
public class Function
{
///
/// Lambda function entry point to process Amazon DocumentDB events.
///
/// The Amazon DocumentDB event.
/// The Lambda context object.
/// A string to indicate successful processing.
public string FunctionHandler(Event evnt, ILambdaContext context)
{
foreach (var record in evnt.Events)
{
ProcessDocumentDBEvent(record, context);
}
return "OK";
}
private void ProcessDocumentDBEvent(DocumentDBEventRecord record, ILambdaContext context)
{
var eventData = record.Event;
var operationType = eventData.OperationType;
var databaseName = eventData.Ns.Db;
var collectionName = eventData.Ns.Coll;
var fullDocument = JsonSerializer.Serialize(eventData.FullDocument, new JsonSerializerOptions { WriteIndented = true });
context.Logger.LogLine($"Operation type: {operationType}");
context.Logger.LogLine($"Database: {databaseName}");
context.Logger.LogLine($"Collection: {collectionName}");
context.Logger.LogLine($"Full document:\n{fullDocument}");
}
public class Event
{
[JsonPropertyName("eventSourceArn")]
public string EventSourceArn { get; set; }
[JsonPropertyName("events")]
public List Events { get; set; }
[JsonPropertyName("eventSource")]
public string EventSource { get; set; }
}
public class DocumentDBEventRecord
{
[JsonPropertyName("event")]
public EventData Event { get; set; }
}
public class EventData
{
[JsonPropertyName("_id")]
public IdData Id { get; set; }
[JsonPropertyName("clusterTime")]
public ClusterTime ClusterTime { get; set; }
[JsonPropertyName("documentKey")]
public DocumentKey DocumentKey { get; set; }
[JsonPropertyName("fullDocument")]
public Dictionary FullDocument { get; set; }
[JsonPropertyName("ns")]
public Namespace Ns { get; set; }
[JsonPropertyName("operationType")]
public string OperationType { get; set; }
}
public class IdData
{
[JsonPropertyName("_data")]
public string Data { get; set; }
}
public class ClusterTime
{
[JsonPropertyName("$timestamp")]
public Timestamp Timestamp { get; set; }
}
public class Timestamp
{
[JsonPropertyName("t")]
public long T { get; set; }
[JsonPropertyName("i")]
public int I { get; set; }
}
public class DocumentKey
{
[JsonPropertyName("_id")]
public Id Id { get; set; }
}
public class Id
{
[JsonPropertyName("$oid")]
public string Oid { get; set; }
}
public class Namespace
{
[JsonPropertyName("db")]
public string Db { get; set; }
[JsonPropertyName("coll")]
public string Coll { get; set; }
}
}
```
------
#### [ Go ]
**适用于 Go 的 SDK V2**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-docdb-to-lambda)存储库中查找完整示例，并了解如何进行设置和运行。
使用 Go 将 Amazon DocumentDB 事件与 Lambda 结合使用。
```
package main
import (
"context"
"encoding/json"
"fmt"
"github.com/aws/aws-lambda-go/lambda"
)
type Event struct {
Events []Record `json:"events"`
}
type Record struct {
Event struct {
OperationType string `json:"operationType"`
NS struct {
DB string `json:"db"`
Coll string `json:"coll"`
} `json:"ns"`
FullDocument interface{} `json:"fullDocument"`
} `json:"event"`
}
func main() {
lambda.Start(handler)
}
func handler(ctx context.Context, event Event) (string, error) {
fmt.Println("Loading function")
for _, record := range event.Events {
logDocumentDBEvent(record)
}
return "OK", nil
}
func logDocumentDBEvent(record Record) {
fmt.Printf("Operation type: %s\n", record.Event.OperationType)
fmt.Printf("db: %s\n", record.Event.NS.DB)
fmt.Printf("collection: %s\n", record.Event.NS.Coll)
docBytes, _ := json.MarshalIndent(record.Event.FullDocument, "", " ")
fmt.Printf("Full document: %s\n", string(docBytes))
}
```
------
#### [ Java ]
**适用于 Java 的 SDK 2.x**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-docdb-to-lambda)存储库中查找完整示例，并了解如何进行设置和运行。
使用 Java 将 Amazon DocumentDB 事件与 Lambda 结合使用。
```
import java.util.List;
import java.util.Map;
import com.amazonaws.services.lambda.runtime.Context;
import com.amazonaws.services.lambda.runtime.RequestHandler;
public class Example implements RequestHandler, String> {
@SuppressWarnings("unchecked")
@Override
public String handleRequest(Map event, Context context) {
List> events = (List>) event.get("events");
for (Map record : events) {
Map eventData = (Map) record.get("event");
processEventData(eventData);
}
return "OK";
}
@SuppressWarnings("unchecked")
private void processEventData(Map eventData) {
String operationType = (String) eventData.get("operationType");
System.out.println("operationType: %s".formatted(operationType));
Map ns = (Map) eventData.get("ns");
String db = (String) ns.get("db");
System.out.println("db: %s".formatted(db));
String coll = (String) ns.get("coll");
System.out.println("coll: %s".formatted(coll));
Map fullDocument = (Map) eventData.get("fullDocument");
System.out.println("fullDocument: %s".formatted(fullDocument));
}
}
```
------
#### [ JavaScript ]
**SDK for JavaScript (v3)**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-docdb-to-lambda)存储库中查找完整示例，并了解如何进行设置和运行。
使用 JavaScript 将 Amazon DocumentDB 事件与 Lambda 结合使用。
```
console.log('Loading function');
exports.handler = async (event, context) => {
event.events.forEach(record => {
logDocumentDBEvent(record);
});
return 'OK';
};
const logDocumentDBEvent = (record) => {
console.log('Operation type: ' + record.event.operationType);
console.log('db: ' + record.event.ns.db);
console.log('collection: ' + record.event.ns.coll);
console.log('Full document:', JSON.stringify(record.event.fullDocument, null, 2));
};
```
使用 TypeScript 将 Amazon DocumentDB 事件与 Lambda 结合使用
```
import { DocumentDBEventRecord, DocumentDBEventSubscriptionContext } from 'aws-lambda';
console.log('Loading function');
export const handler = async (
event: DocumentDBEventSubscriptionContext,
context: any
): Promise => {
event.events.forEach((record: DocumentDBEventRecord) => {
logDocumentDBEvent(record);
});
return 'OK';
};
const logDocumentDBEvent = (record: DocumentDBEventRecord): void => {
console.log('Operation type: ' + record.event.operationType);
console.log('db: ' + record.event.ns.db);
console.log('collection: ' + record.event.ns.coll);
console.log('Full document:', JSON.stringify(record.event.fullDocument, null, 2));
};
```
------
#### [ PHP ]
**适用于 PHP 的 SDK**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-docdb-to-lambda)存储库中查找完整示例，并了解如何进行设置和运行。
使用 PHP 将 Amazon DocumentDB 事件与 Lambda 结合使用。
```
logDocumentDBEvent($record['event']);
}
return 'OK';
}
private function logDocumentDBEvent($event): void
{
// Extract information from the event record
$operationType = $event['operationType'] ?? 'Unknown';
$db = $event['ns']['db'] ?? 'Unknown';
$collection = $event['ns']['coll'] ?? 'Unknown';
$fullDocument = $event['fullDocument'] ?? [];
// Log the event details
echo "Operation type: $operationType\n";
echo "Database: $db\n";
echo "Collection: $collection\n";
echo "Full document: " . json_encode($fullDocument, JSON_PRETTY_PRINT) . "\n";
}
}
return new DocumentDBEventHandler();
```
------
#### [ Python ]
**适用于 Python 的 SDK（Boto3）**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-docdb-to-lambda)存储库中查找完整示例，并了解如何进行设置和运行。
使用 Python 将 Amazon DocumentDB 事件与 Lambda 结合使用。
```
import json
def lambda_handler(event, context):
for record in event.get('events', []):
log_document_db_event(record)
return 'OK'
def log_document_db_event(record):
event_data = record.get('event', {})
operation_type = event_data.get('operationType', 'Unknown')
db = event_data.get('ns', {}).get('db', 'Unknown')
collection = event_data.get('ns', {}).get('coll', 'Unknown')
full_document = event_data.get('fullDocument', {})
print(f"Operation type: {operation_type}")
print(f"db: {db}")
print(f"collection: {collection}")
print("Full document:", json.dumps(full_document, indent=2))
```
------
#### [ Ruby ]
**适用于 Ruby 的 SDK**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-docdb-to-lambda)存储库中查找完整示例，并了解如何进行设置和运行。
使用 Ruby 将 Amazon DocumentDB 事件与 Lambda 结合使用。
```
require 'json'
def lambda_handler(event:, context:)
event['events'].each do |record|
log_document_db_event(record)
end
'OK'
end
def log_document_db_event(record)
event_data = record['event'] || {}
operation_type = event_data['operationType'] || 'Unknown'
db = event_data.dig('ns', 'db') || 'Unknown'
collection = event_data.dig('ns', 'coll') || 'Unknown'
full_document = event_data['fullDocument'] || {}
puts "Operation type: #{operation_type}"
puts "db: #{db}"
puts "collection: #{collection}"
puts "Full document: #{JSON.pretty_generate(full_document)}"
end
```
------
#### [ Rust ]
**适用于 Rust 的 SDK**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-docdb-to-lambda)存储库中查找完整示例，并了解如何进行设置和运行。
使用 Rust 将 Amazon DocumentDB 事件与 Lambda 结合使用。
```
use lambda_runtime::{service_fn, tracing, Error, LambdaEvent};
use aws_lambda_events::{
event::documentdb::{DocumentDbEvent, DocumentDbInnerEvent},
};
// 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) ->Result<(), Error> {
tracing::info!("Event Source ARN: {:?}", event.payload.event_source_arn);
tracing::info!("Event Source: {:?}", event.payload.event_source);
let records = &event.payload.events;
if records.is_empty() {
tracing::info!("No records found. Exiting.");
return Ok(());
}
for record in records{
log_document_db_event(record);
}
tracing::info!("Document db records processed");
// Prepare the response
Ok(())
}
fn log_document_db_event(record: &DocumentDbInnerEvent)-> Result<(), Error>{
tracing::info!("Change Event: {:?}", record.event);
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(())
}
```
------
有关 AWS SDK 开发人员指南和代码示例的完整列表，请参阅 [将 Lambda 与 AWS SDK 配合使用](sdk-general-information-section.md) 本主题还包括有关入门的信息以及有关先前的 SDK 版本的详细信息。
# 通过 Amazon MSK 触发器调用 Lambda 函数
以下代码示例展示了如何实现 Lambda 函数，该函数接收通过接收来自 Amazon MSK 集群的记录而触发的事件。该函数检索 MSK 有效负载，并记录下记录内容。
------
#### [ .NET ]
**适用于 .NET 的 SDK**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-msk-to-lambda)存储库中查找完整示例，并了解如何进行设置和运行。
通过 .NET 将 Amazon MSK 事件与 Lambda 结合使用。
```
using System.Text;
using Amazon.Lambda.Core;
using Amazon.Lambda.KafkaEvents;
// 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 MSKLambda;
public class Function
{
/// The event for the Lambda function handler to process.
/// The ILambdaContext that provides methods for logging and describing the Lambda environment.
///
public void FunctionHandler(KafkaEvent evnt, ILambdaContext context)
{
foreach (var record in evnt.Records)
{
Console.WriteLine("Key:" + record.Key);
foreach (var eventRecord in record.Value)
{
var valueBytes = eventRecord.Value.ToArray();
var valueText = Encoding.UTF8.GetString(valueBytes);
Console.WriteLine("Message:" + valueText);
}
}
}
}
```
------
#### [ Go ]
**适用于 Go 的 SDK V2**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-msk-to-lambda)存储库中查找完整示例，并了解如何进行设置和运行。
通过 Go 将 Amazon MSK 事件与 Lambda 结合使用。
```
package main
import (
"encoding/base64"
"fmt"
"github.com/aws/aws-lambda-go/events"
"github.com/aws/aws-lambda-go/lambda"
)
func handler(event events.KafkaEvent) {
for key, records := range event.Records {
fmt.Println("Key:", key)
for _, record := range records {
fmt.Println("Record:", record)
decodedValue, _ := base64.StdEncoding.DecodeString(record.Value)
message := string(decodedValue)
fmt.Println("Message:", message)
}
}
}
func main() {
lambda.Start(handler)
}
```
------
#### [ Java ]
**适用于 Java 的 SDK 2.x**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-msk-to-lambda)存储库中查找完整示例，并了解如何进行设置和运行。
通过 Java 将 Amazon MSK 事件与 Lambda 结合使用。
```
import com.amazonaws.services.lambda.runtime.Context;
import com.amazonaws.services.lambda.runtime.RequestHandler;
import com.amazonaws.services.lambda.runtime.events.KafkaEvent;
import com.amazonaws.services.lambda.runtime.events.KafkaEvent.KafkaEventRecord;
import java.util.Base64;
import java.util.Map;
public class Example implements RequestHandler {
@Override
public Void handleRequest(KafkaEvent event, Context context) {
for (Map.Entry> entry : event.getRecords().entrySet()) {
String key = entry.getKey();
System.out.println("Key: " + key);
for (KafkaEventRecord record : entry.getValue()) {
System.out.println("Record: " + record);
byte[] value = Base64.getDecoder().decode(record.getValue());
String message = new String(value);
System.out.println("Message: " + message);
}
}
return null;
}
}
```
------
#### [ JavaScript ]
**SDK for JavaScript（v3）**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-msk-to-lambda)存储库中查找完整示例，并了解如何进行设置和运行。
通过 JavaScript 将 Amazon MSK 事件与 Lambda 结合使用。
```
exports.handler = async (event) => {
// Iterate through keys
for (let key in event.records) {
console.log('Key: ', key)
// Iterate through records
event.records[key].map((record) => {
console.log('Record: ', record)
// Decode base64
const msg = Buffer.from(record.value, 'base64').toString()
console.log('Message:', msg)
})
}
}
```
通过 TypeScript 将 Amazon MSK 事件与 Lambda 结合使用。
```
import { MSKEvent, Context } from "aws-lambda";
import { Buffer } from "buffer";
import { Logger } from "@aws-lambda-powertools/logger";
const logger = new Logger({
logLevel: "INFO",
serviceName: "msk-handler-sample",
});
export const handler = async (
event: MSKEvent,
context: Context
): Promise => {
for (const [topic, topicRecords] of Object.entries(event.records)) {
logger.info(`Processing key: ${topic}`);
// Process each record in the partition
for (const record of topicRecords) {
try {
// Decode the message value from base64
const decodedMessage = Buffer.from(record.value, 'base64').toString();
logger.info({
message: decodedMessage
});
}
catch (error) {
logger.error('Error processing event', { error });
throw error;
}
};
}
}
```
------
#### [ PHP ]
**适用于 PHP 的 SDK**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-msk-to-lambda)存储库中查找完整示例，并了解如何进行设置和运行。
通过 PHP 将 Amazon MSK 事件与 Lambda 结合使用。
```
logger = $logger;
}
/**
* @throws JsonException
* @throws \Bref\Event\InvalidLambdaEvent
*/
public function handle(mixed $event, Context $context): void
{
$kafkaEvent = new KafkaEvent($event);
$this->logger->info("Processing records");
$records = $kafkaEvent->getRecords();
foreach ($records as $record) {
try {
$key = $record->getKey();
$this->logger->info("Key: $key");
$values = $record->getValue();
$this->logger->info(json_encode($values));
foreach ($values as $value) {
$this->logger->info("Value: $value");
}
} catch (Exception $e) {
$this->logger->error($e->getMessage());
}
}
$totalRecords = count($records);
$this->logger->info("Successfully processed $totalRecords records");
}
}
$logger = new StderrLogger();
return new Handler($logger);
```
------
#### [ Python ]
**适用于 Python 的 SDK（Boto3）**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-msk-to-lambda)存储库中查找完整示例，并了解如何进行设置和运行。
通过 Python 将 Amazon MSK 事件与 Lambda 结合使用。
```
import base64
def lambda_handler(event, context):
# Iterate through keys
for key in event['records']:
print('Key:', key)
# Iterate through records
for record in event['records'][key]:
print('Record:', record)
# Decode base64
msg = base64.b64decode(record['value']).decode('utf-8')
print('Message:', msg)
```
------
#### [ Ruby ]
**适用于 Ruby 的 SDK**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-msk-to-lambda)存储库中查找完整示例，并了解如何进行设置和运行。
通过 Ruby 将 Amazon MSK 事件与 Lambda 结合使用。
```
require 'base64'
def lambda_handler(event:, context:)
# Iterate through keys
event['records'].each do |key, records|
puts "Key: #{key}"
# Iterate through records
records.each do |record|
puts "Record: #{record}"
# Decode base64
msg = Base64.decode64(record['value'])
puts "Message: #{msg}"
end
end
end
```
------
#### [ Rust ]
**适用于 Rust 的 SDK**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-msk-to-lambda)存储库中查找完整示例，并了解如何进行设置和运行。
通过 Rust 将 Amazon MSK 事件与 Lambda 结合使用。
```
use aws_lambda_events::event::kafka::KafkaEvent;
use lambda_runtime::{run, service_fn, tracing, Error, LambdaEvent};
use base64::prelude::*;
use serde_json::{Value};
use tracing::{info};
/// Pre-Requisites:
/// 1. Install Cargo Lambda - see https://www.cargo-lambda.info/guide/getting-started.html
/// 2. Add packages tracing, tracing-subscriber, serde_json, base64
///
/// This is the main body for the function.
/// Write your code inside it.
/// There are some code example in the following URLs:
/// - https://github.com/awslabs/aws-lambda-rust-runtime/tree/main/examples
/// - https://github.com/aws-samples/serverless-rust-demo/
async fn function_handler(event: LambdaEvent) -> Result {
let payload = event.payload.records;
for (_name, records) in payload.iter() {
for record in records {
let record_text = record.value.as_ref().ok_or("Value is None")?;
info!("Record: {}", &record_text);
// perform Base64 decoding
let record_bytes = BASE64_STANDARD.decode(record_text)?;
let message = std::str::from_utf8(&record_bytes)?;
info!("Message: {}", message);
}
}
Ok(().into())
}
#[tokio::main]
async fn main() -> Result<(), Error> {
// required to enable CloudWatch error logging by the runtime
tracing::init_default_subscriber();
info!("Setup CW subscriber!");
run(service_fn(function_handler)).await
}
```
------
有关 AWS SDK 开发人员指南和代码示例的完整列表，请参阅 [将 Lambda 与 AWS SDK 配合使用](sdk-general-information-section.md) 本主题还包括有关入门的信息以及有关先前的 SDK 版本的详细信息。
# 通过 Amazon S3 触发器调用 Lambda 函数
以下代码示例演示了如何实现一个 Lambda 函数，该函数接收通过将对象上传到 S3 存储桶而触发的事件。该函数从事件参数中检索 S3 存储桶名称和对象密钥，并调用 Amazon S3 API 来检索和记录对象的内容类型。
------
#### [ .NET ]
**适用于 .NET 的 SDK**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-s3-to-lambda)存储库中查找完整示例，并了解如何进行设置和运行。
使用 .NET 将 S3 事件与 Lambda 结合使用。
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
using System.Threading.Tasks;
using Amazon.Lambda.Core;
using Amazon.S3;
using System;
using Amazon.Lambda.S3Events;
using System.Web;
// 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 S3Integration
{
public class Function
{
private static AmazonS3Client _s3Client;
public Function() : this(null)
{
}
internal Function(AmazonS3Client s3Client)
{
_s3Client = s3Client ?? new AmazonS3Client();
}
public async Task Handler(S3Event evt, ILambdaContext context)
{
try
{
if (evt.Records.Count <= 0)
{
context.Logger.LogLine("Empty S3 Event received");
return string.Empty;
}
var bucket = evt.Records[0].S3.Bucket.Name;
var key = HttpUtility.UrlDecode(evt.Records[0].S3.Object.Key);
context.Logger.LogLine($"Request is for {bucket} and {key}");
var objectResult = await _s3Client.GetObjectAsync(bucket, key);
context.Logger.LogLine($"Returning {objectResult.Key}");
return objectResult.Key;
}
catch (Exception e)
{
context.Logger.LogLine($"Error processing request - {e.Message}");
return string.Empty;
}
}
}
}
```
------
#### [ Go ]
**适用于 Go 的 SDK V2**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-s3-to-lambda)存储库中查找完整示例，并了解如何进行设置和运行。
使用 Go 将 S3 事件与 Lambda 结合使用。
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
package main
import (
"context"
"log"
"github.com/aws/aws-lambda-go/events"
"github.com/aws/aws-lambda-go/lambda"
"github.com/aws/aws-sdk-go-v2/config"
"github.com/aws/aws-sdk-go-v2/service/s3"
)
func handler(ctx context.Context, s3Event events.S3Event) error {
sdkConfig, err := config.LoadDefaultConfig(ctx)
if err != nil {
log.Printf("failed to load default config: %s", err)
return err
}
s3Client := s3.NewFromConfig(sdkConfig)
for _, record := range s3Event.Records {
bucket := record.S3.Bucket.Name
key := record.S3.Object.URLDecodedKey
headOutput, err := s3Client.HeadObject(ctx, &s3.HeadObjectInput{
Bucket: &bucket,
Key: &key,
})
if err != nil {
log.Printf("error getting head of object %s/%s: %s", bucket, key, err)
return err
}
log.Printf("successfully retrieved %s/%s of type %s", bucket, key, *headOutput.ContentType)
}
return nil
}
func main() {
lambda.Start(handler)
}
```
------
#### [ Java ]
**适用于 Java 的 SDK 2.x**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-s3-to-lambda)存储库中查找完整示例，并了解如何进行设置和运行。
使用 Java 将 S3 事件与 Lambda 结合使用。
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
package example;
import software.amazon.awssdk.services.s3.model.HeadObjectRequest;
import software.amazon.awssdk.services.s3.model.HeadObjectResponse;
import software.amazon.awssdk.services.s3.S3Client;
import com.amazonaws.services.lambda.runtime.Context;
import com.amazonaws.services.lambda.runtime.RequestHandler;
import com.amazonaws.services.lambda.runtime.events.S3Event;
import com.amazonaws.services.lambda.runtime.events.models.s3.S3EventNotification.S3EventNotificationRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class Handler implements RequestHandler {
private static final Logger logger = LoggerFactory.getLogger(Handler.class);
@Override
public String handleRequest(S3Event s3event, Context context) {
try {
S3EventNotificationRecord record = s3event.getRecords().get(0);
String srcBucket = record.getS3().getBucket().getName();
String srcKey = record.getS3().getObject().getUrlDecodedKey();
S3Client s3Client = S3Client.builder().build();
HeadObjectResponse headObject = getHeadObject(s3Client, srcBucket, srcKey);
logger.info("Successfully retrieved " + srcBucket + "/" + srcKey + " of type " + headObject.contentType());
return "Ok";
} catch (Exception e) {
throw new RuntimeException(e);
}
}
private HeadObjectResponse getHeadObject(S3Client s3Client, String bucket, String key) {
HeadObjectRequest headObjectRequest = HeadObjectRequest.builder()
.bucket(bucket)
.key(key)
.build();
return s3Client.headObject(headObjectRequest);
}
}
```
------
#### [ JavaScript ]
**SDK for JavaScript（v3）**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-s3-to-lambda)存储库中查找完整示例，并了解如何进行设置和运行。
使用 JavaScript 将 S3 事件与 Lambda 结合使用。
```
import { S3Client, HeadObjectCommand } from "@aws-sdk/client-s3";
const client = new S3Client();
export const handler = async (event, context) => {
// Get the object from the event and show its content type
const bucket = event.Records[0].s3.bucket.name;
const key = decodeURIComponent(event.Records[0].s3.object.key.replace(/\+/g, ' '));
try {
const { ContentType } = await client.send(new HeadObjectCommand({
Bucket: bucket,
Key: key,
}));
console.log('CONTENT TYPE:', ContentType);
return ContentType;
} catch (err) {
console.log(err);
const message = `Error getting object ${key} from bucket ${bucket}. Make sure they exist and your bucket is in the same region as this function.`;
console.log(message);
throw new Error(message);
}
};
```
使用 TypeScript 将 S3 事件与 Lambda 结合使用。
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
import { S3Event } from 'aws-lambda';
import { S3Client, HeadObjectCommand } from '@aws-sdk/client-s3';
const s3 = new S3Client({ region: process.env.AWS_REGION });
export const handler = async (event: S3Event): Promise => {
// Get the object from the event and show its content type
const bucket = event.Records[0].s3.bucket.name;
const key = decodeURIComponent(event.Records[0].s3.object.key.replace(/\+/g, ' '));
const params = {
Bucket: bucket,
Key: key,
};
try {
const { ContentType } = await s3.send(new HeadObjectCommand(params));
console.log('CONTENT TYPE:', ContentType);
return ContentType;
} catch (err) {
console.log(err);
const message = `Error getting object ${key} from bucket ${bucket}. Make sure they exist and your bucket is in the same region as this function.`;
console.log(message);
throw new Error(message);
}
};
```
------
#### [ PHP ]
**适用于 PHP 的 SDK**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-s3-to-lambda)存储库中查找完整示例，并了解如何进行设置和运行。
通过 PHP 将 S3 事件与 Lambda 结合使用。
```
logger = $logger;
}
public function handleS3(S3Event $event, Context $context) : void
{
$this->logger->info("Processing S3 records");
// Get the object from the event and show its content type
$records = $event->getRecords();
foreach ($records as $record)
{
$bucket = $record->getBucket()->getName();
$key = urldecode($record->getObject()->getKey());
try {
$fileSize = urldecode($record->getObject()->getSize());
echo "File Size: " . $fileSize . "\n";
// TODO: Implement your custom processing logic here
} catch (Exception $e) {
echo $e->getMessage() . "\n";
echo 'Error getting object ' . $key . ' from bucket ' . $bucket . '. Make sure they exist and your bucket is in the same region as this function.' . "\n";
throw $e;
}
}
}
}
$logger = new StderrLogger();
return new Handler($logger);
```
------
#### [ Python ]
**适用于 Python 的 SDK（Boto3）**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-s3-to-lambda)存储库中查找完整示例，并了解如何进行设置和运行。
使用 Python 将 S3 事件与 Lambda 结合使用。
```
# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
# SPDX-License-Identifier: Apache-2.0
import json
import urllib.parse
import boto3
print('Loading function')
s3 = boto3.client('s3')
def lambda_handler(event, context):
#print("Received event: " + json.dumps(event, indent=2))
# Get the object from the event and show its content type
bucket = event['Records'][0]['s3']['bucket']['name']
key = urllib.parse.unquote_plus(event['Records'][0]['s3']['object']['key'], encoding='utf-8')
try:
response = s3.get_object(Bucket=bucket, Key=key)
print("CONTENT TYPE: " + response['ContentType'])
return response['ContentType']
except Exception as e:
print(e)
print('Error getting object {} from bucket {}. Make sure they exist and your bucket is in the same region as this function.'.format(key, bucket))
raise e
```
------
#### [ Ruby ]
**适用于 Ruby 的 SDK**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-s3-to-lambda)存储库中查找完整示例，并了解如何进行设置和运行。
通过 Ruby 将 S3 事件与 Lambda 结合使用。
```
require 'json'
require 'uri'
require 'aws-sdk'
puts 'Loading function'
def lambda_handler(event:, context:)
s3 = Aws::S3::Client.new(region: 'region') # Your AWS region
# puts "Received event: #{JSON.dump(event)}"
# Get the object from the event and show its content type
bucket = event['Records'][0]['s3']['bucket']['name']
key = URI.decode_www_form_component(event['Records'][0]['s3']['object']['key'], Encoding::UTF_8)
begin
response = s3.get_object(bucket: bucket, key: key)
puts "CONTENT TYPE: #{response.content_type}"
return response.content_type
rescue StandardError => e
puts e.message
puts "Error getting object #{key} from bucket #{bucket}. Make sure they exist and your bucket is in the same region as this function."
raise e
end
end
```
------
#### [ Rust ]
**适用于 Rust 的 SDK**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-s3-to-lambda)存储库中查找完整示例，并了解如何进行设置和运行。
使用 Rust 将 S3 事件与 Lambda 结合使用。
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
use aws_lambda_events::event::s3::S3Event;
use aws_sdk_s3::{Client};
use lambda_runtime::{run, service_fn, Error, LambdaEvent};
/// Main function
#[tokio::main]
async fn main() -> Result<(), Error> {
tracing_subscriber::fmt()
.with_max_level(tracing::Level::INFO)
.with_target(false)
.without_time()
.init();
// Initialize the AWS SDK for Rust
let config = aws_config::load_from_env().await;
let s3_client = Client::new(&config);
let res = run(service_fn(|request: LambdaEvent| {
function_handler(&s3_client, request)
})).await;
res
}
async fn function_handler(
s3_client: &Client,
evt: LambdaEvent
) -> Result<(), Error> {
tracing::info!(records = ?evt.payload.records.len(), "Received request from SQS");
if evt.payload.records.len() == 0 {
tracing::info!("Empty S3 event received");
}
let bucket = evt.payload.records[0].s3.bucket.name.as_ref().expect("Bucket name to exist");
let key = evt.payload.records[0].s3.object.key.as_ref().expect("Object key to exist");
tracing::info!("Request is for {} and object {}", bucket, key);
let s3_get_object_result = s3_client
.get_object()
.bucket(bucket)
.key(key)
.send()
.await;
match s3_get_object_result {
Ok(_) => tracing::info!("S3 Get Object success, the s3GetObjectResult contains a 'body' property of type ByteStream"),
Err(_) => tracing::info!("Failure with S3 Get Object request")
}
Ok(())
}
```
------
有关 AWS SDK 开发人员指南和代码示例的完整列表，请参阅 [将 Lambda 与 AWS SDK 配合使用](sdk-general-information-section.md) 本主题还包括有关入门的信息以及有关先前的 SDK 版本的详细信息。
# 通过 Amazon SNS 触发器调用 Lambda 函数
以下代码示例演示了如何实现一个 Lambda 函数，该函数接收通过接收来自 SNS 主题的消息而触发的事件。该函数从事件参数检索消息并记录每条消息的内容。
------
#### [ .NET ]
**适用于 .NET 的 SDK**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-sns-to-lambda)存储库中查找完整示例，并了解如何进行设置和运行。
使用 .NET 将 SNS 事件与 Lambda 结合使用。
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
using Amazon.Lambda.Core;
using Amazon.Lambda.SNSEvents;
// 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 SnsIntegration;
public class Function
{
public async Task FunctionHandler(SNSEvent evnt, ILambdaContext context)
{
foreach (var record in evnt.Records)
{
await ProcessRecordAsync(record, context);
}
context.Logger.LogInformation("done");
}
private async Task ProcessRecordAsync(SNSEvent.SNSRecord record, ILambdaContext context)
{
try
{
context.Logger.LogInformation($"Processed record {record.Sns.Message}");
// TODO: Do interesting work based on the new message
await Task.CompletedTask;
}
catch (Exception e)
{
//You can use Dead Letter Queue to handle failures. By configuring a Lambda DLQ.
context.Logger.LogError($"An error occurred");
throw;
}
}
}
```
------
#### [ Go ]
**适用于 Go 的 SDK V2**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-sns-to-lambda)存储库中查找完整示例，并了解如何进行设置和运行。
使用 Go 将 SNS 事件与 Lambda 结合使用。
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
package main
import (
"context"
"fmt"
"github.com/aws/aws-lambda-go/events"
"github.com/aws/aws-lambda-go/lambda"
)
func handler(ctx context.Context, snsEvent events.SNSEvent) {
for _, record := range snsEvent.Records {
processMessage(record)
}
fmt.Println("done")
}
func processMessage(record events.SNSEventRecord) {
message := record.SNS.Message
fmt.Printf("Processed message: %s\n", message)
// TODO: Process your record here
}
func main() {
lambda.Start(handler)
}
```
------
#### [ Java ]
**适用于 Java 的 SDK 2.x**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-sns-to-lambda)存储库中查找完整示例，并了解如何进行设置和运行。
通过 Java 将 SNS 事件与 Lambda 结合使用。
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
package example;
import com.amazonaws.services.lambda.runtime.Context;
import com.amazonaws.services.lambda.runtime.LambdaLogger;
import com.amazonaws.services.lambda.runtime.RequestHandler;
import com.amazonaws.services.lambda.runtime.events.SNSEvent;
import com.amazonaws.services.lambda.runtime.events.SNSEvent.SNSRecord;
import java.util.Iterator;
import java.util.List;
public class SNSEventHandler implements RequestHandler {
LambdaLogger logger;
@Override
public Boolean handleRequest(SNSEvent event, Context context) {
logger = context.getLogger();
List records = event.getRecords();
if (!records.isEmpty()) {
Iterator recordsIter = records.iterator();
while (recordsIter.hasNext()) {
processRecord(recordsIter.next());
}
}
return Boolean.TRUE;
}
public void processRecord(SNSRecord record) {
try {
String message = record.getSNS().getMessage();
logger.log("message: " + message);
} catch (Exception e) {
throw new RuntimeException(e);
}
}
}
```
------
#### [ JavaScript ]
**SDK for JavaScript（v3）**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/blob/main/integration-sns-to-lambda)存储库中查找完整示例，并了解如何进行设置和运行。
使用 JavaScript 将 SNS 事件与 Lambda 结合使用。
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
exports.handler = async (event, context) => {
for (const record of event.Records) {
await processMessageAsync(record);
}
console.info("done");
};
async function processMessageAsync(record) {
try {
const message = JSON.stringify(record.Sns.Message);
console.log(`Processed message ${message}`);
await Promise.resolve(1); //Placeholder for actual async work
} catch (err) {
console.error("An error occurred");
throw err;
}
}
```
使用 TypeScript 将 SNS 事件与 Lambda 结合使用。
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
import { SNSEvent, Context, SNSHandler, SNSEventRecord } from "aws-lambda";
export const functionHandler: SNSHandler = async (
event: SNSEvent,
context: Context
): Promise => {
for (const record of event.Records) {
await processMessageAsync(record);
}
console.info("done");
};
async function processMessageAsync(record: SNSEventRecord): Promise {
try {
const message: string = JSON.stringify(record.Sns.Message);
console.log(`Processed message ${message}`);
await Promise.resolve(1); //Placeholder for actual async work
} catch (err) {
console.error("An error occurred");
throw err;
}
}
```
------
#### [ PHP ]
**适用于 PHP 的 SDK**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-sns-to-lambda)存储库中查找完整示例，并了解如何进行设置和运行。
通过 PHP 将 SNS 事件与 Lambda 结合使用。
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
getRecords() as $record) {
$message = $record->getMessage();
// TODO: Implement your custom processing logic here
// Any exception thrown will be logged and the invocation will be marked as failed
echo "Processed Message: $message" . PHP_EOL;
}
}
}
return new Handler();
```
------
#### [ Python ]
**适用于 Python 的 SDK（Boto3）**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-sns-to-lambda)存储库中查找完整示例，并了解如何进行设置和运行。
使用 Python 将 SNS 事件与 Lambda 结合使用。
```
# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
# SPDX-License-Identifier: Apache-2.0
def lambda_handler(event, context):
for record in event['Records']:
process_message(record)
print("done")
def process_message(record):
try:
message = record['Sns']['Message']
print(f"Processed message {message}")
# TODO; Process your record here
except Exception as e:
print("An error occurred")
raise e
```
------
#### [ Ruby ]
**适用于 Ruby 的 SDK**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-sns-to-lambda)存储库中查找完整示例，并了解如何进行设置和运行。
通过 Ruby 将 SNS 事件与 Lambda 结合使用。
```
# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
# SPDX-License-Identifier: Apache-2.0
def lambda_handler(event:, context:)
event['Records'].map { |record| process_message(record) }
end
def process_message(record)
message = record['Sns']['Message']
puts("Processing message: #{message}")
rescue StandardError => e
puts("Error processing message: #{e}")
raise
end
```
------
#### [ Rust ]
**适用于 Rust 的 SDK**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-sns-to-lambda)存储库中查找完整示例，并了解如何进行设置和运行。
使用 Rust 将 SNS 事件与 Lambda 结合使用。
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
use aws_lambda_events::event::sns::SnsEvent;
use aws_lambda_events::sns::SnsRecord;
use lambda_runtime::{run, service_fn, Error, LambdaEvent};
use tracing::info;
// Built with the following dependencies:
// aws_lambda_events = { version = "0.10.0", default-features = false, features = ["sns"] }
// lambda_runtime = "0.8.1"
// tokio = { version = "1", features = ["macros"] }
// tracing = { version = "0.1", features = ["log"] }
// tracing-subscriber = { version = "0.3", default-features = false, features = ["fmt"] }
async fn function_handler(event: LambdaEvent) -> Result<(), Error> {
for event in event.payload.records {
process_record(&event)?;
}
Ok(())
}
fn process_record(record: &SnsRecord) -> Result<(), Error> {
info!("Processing SNS Message: {}", record.sns.message);
// Implement your record handling code here.
Ok(())
}
#[tokio::main]
async fn main() -> Result<(), Error> {
tracing_subscriber::fmt()
.with_max_level(tracing::Level::INFO)
.with_target(false)
.without_time()
.init();
run(service_fn(function_handler)).await
}
```
------
有关 AWS SDK 开发人员指南和代码示例的完整列表，请参阅 [将 Lambda 与 AWS SDK 配合使用](sdk-general-information-section.md) 本主题还包括有关入门的信息以及有关先前的 SDK 版本的详细信息。
# 通过 Amazon SQS 触发器调用 Lambda 函数
以下代码示例演示了如何实现一个 Lambda 函数，该函数接收通过接收来自 SQS 队列的消息而触发的事件。该函数从事件参数检索消息并记录每条消息的内容。
------
#### [ .NET ]
**适用于 .NET 的 SDK**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-sqs-to-lambda)存储库中查找完整示例，并了解如何进行设置和运行。
通过 .NET 将 SQS 事件与 Lambda 结合使用。
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
using Amazon.Lambda.Core;
using Amazon.Lambda.SQSEvents;
// 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 SqsIntegrationSampleCode
{
public async Task FunctionHandler(SQSEvent evnt, ILambdaContext context)
{
foreach (var message in evnt.Records)
{
await ProcessMessageAsync(message, context);
}
context.Logger.LogInformation("done");
}
private async Task ProcessMessageAsync(SQSEvent.SQSMessage message, ILambdaContext context)
{
try
{
context.Logger.LogInformation($"Processed message {message.Body}");
// TODO: Do interesting work based on the new message
await Task.CompletedTask;
}
catch (Exception e)
{
//You can use Dead Letter Queue to handle failures. By configuring a Lambda DLQ.
context.Logger.LogError($"An error occurred");
throw;
}
}
}
```
------
#### [ Go ]
**适用于 Go 的 SDK V2**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-sqs-to-lambda)存储库中查找完整示例，并了解如何进行设置和运行。
使用 Go 将 SQS 事件与 Lambda 结合使用。
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
package integration_sqs_to_lambda
import (
"fmt"
"github.com/aws/aws-lambda-go/events"
"github.com/aws/aws-lambda-go/lambda"
)
func handler(event events.SQSEvent) error {
for _, record := range event.Records {
err := processMessage(record)
if err != nil {
return err
}
}
fmt.Println("done")
return nil
}
func processMessage(record events.SQSMessage) error {
fmt.Printf("Processed message %s\n", record.Body)
// TODO: Do interesting work based on the new message
return nil
}
func main() {
lambda.Start(handler)
}
```
------
#### [ Java ]
**适用于 Java 的 SDK 2.x**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-sqs-to-lambda)存储库中查找完整示例，并了解如何进行设置和运行。
通过 Java 将 SQS 事件与 Lambda 结合使用。
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
import com.amazonaws.services.lambda.runtime.Context;
import com.amazonaws.services.lambda.runtime.RequestHandler;
import com.amazonaws.services.lambda.runtime.events.SQSEvent;
import com.amazonaws.services.lambda.runtime.events.SQSEvent.SQSMessage;
public class Function implements RequestHandler {
@Override
public Void handleRequest(SQSEvent sqsEvent, Context context) {
for (SQSMessage msg : sqsEvent.getRecords()) {
processMessage(msg, context);
}
context.getLogger().log("done");
return null;
}
private void processMessage(SQSMessage msg, Context context) {
try {
context.getLogger().log("Processed message " + msg.getBody());
// TODO: Do interesting work based on the new message
} catch (Exception e) {
context.getLogger().log("An error occurred");
throw e;
}
}
}
```
------
#### [ JavaScript ]
**SDK for JavaScript（v3）**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/blob/main/integration-sqs-to-lambda)存储库中查找完整示例，并了解如何进行设置和运行。
通过 JavaScript 将 SQS 事件与 Lambda 结合使用。
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
exports.handler = async (event, context) => {
for (const message of event.Records) {
await processMessageAsync(message);
}
console.info("done");
};
async function processMessageAsync(message) {
try {
console.log(`Processed message ${message.body}`);
// TODO: Do interesting work based on the new message
await Promise.resolve(1); //Placeholder for actual async work
} catch (err) {
console.error("An error occurred");
throw err;
}
}
```
通过 TypeScript 将 SQS 事件与 Lambda 结合使用。
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
import { SQSEvent, Context, SQSHandler, SQSRecord } from "aws-lambda";
export const functionHandler: SQSHandler = async (
event: SQSEvent,
context: Context
): Promise => {
for (const message of event.Records) {
await processMessageAsync(message);
}
console.info("done");
};
async function processMessageAsync(message: SQSRecord): Promise {
try {
console.log(`Processed message ${message.body}`);
// TODO: Do interesting work based on the new message
await Promise.resolve(1); //Placeholder for actual async work
} catch (err) {
console.error("An error occurred");
throw err;
}
}
```
------
#### [ PHP ]
**适用于 PHP 的 SDK**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-sqs-to-lambda)存储库中查找完整示例，并了解如何进行设置和运行。
使用 PHP 将 SQS 事件与 Lambda 结合使用。
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
logger = $logger;
}
/**
* @throws InvalidLambdaEvent
*/
public function handleSqs(SqsEvent $event, Context $context): void
{
foreach ($event->getRecords() as $record) {
$body = $record->getBody();
// TODO: Do interesting work based on the new message
}
}
}
$logger = new StderrLogger();
return new Handler($logger);
```
------
#### [ Python ]
**适用于 Python 的 SDK（Boto3）**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-sqs-to-lambda)存储库中查找完整示例，并了解如何进行设置和运行。
使用 Python 将 SQS 事件与 Lambda 结合使用。
```
# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
# SPDX-License-Identifier: Apache-2.0
def lambda_handler(event, context):
for message in event['Records']:
process_message(message)
print("done")
def process_message(message):
try:
print(f"Processed message {message['body']}")
# TODO: Do interesting work based on the new message
except Exception as err:
print("An error occurred")
raise err
```
------
#### [ Ruby ]
**适用于 Ruby 的 SDK**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-sqs-to-lambda)存储库中查找完整示例，并了解如何进行设置和运行。
使用 Ruby 将 SQS 事件与 Lambda 结合使用。
```
# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
# SPDX-License-Identifier: Apache-2.0
def lambda_handler(event:, context:)
event['Records'].each do |message|
process_message(message)
end
puts "done"
end
def process_message(message)
begin
puts "Processed message #{message['body']}"
# TODO: Do interesting work based on the new message
rescue StandardError => err
puts "An error occurred"
raise err
end
end
```
------
#### [ Rust ]
**适用于 Rust 的 SDK**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-sqs-to-lambda)存储库中查找完整示例，并了解如何进行设置和运行。
通过 Rust 将 SQS 事件与 Lambda 结合使用。
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
use aws_lambda_events::event::sqs::SqsEvent;
use lambda_runtime::{run, service_fn, Error, LambdaEvent};
async fn function_handler(event: LambdaEvent) -> Result<(), Error> {
event.payload.records.iter().for_each(|record| {
// process the record
tracing::info!("Message body: {}", record.body.as_deref().unwrap_or_default())
});
Ok(())
}
#[tokio::main]
async fn main() -> Result<(), Error> {
tracing_subscriber::fmt()
.with_max_level(tracing::Level::INFO)
// disable printing the name of the module in every log line.
.with_target(false)
// disabling time is handy because CloudWatch will add the ingestion time.
.without_time()
.init();
run(service_fn(function_handler)).await
}
```
------
有关 AWS SDK 开发人员指南和代码示例的完整列表，请参阅 [将 Lambda 与 AWS SDK 配合使用](sdk-general-information-section.md) 本主题还包括有关入门的信息以及有关先前的 SDK 版本的详细信息。
# 通过 Kinesis 触发器报告 Lambda 函数批处理项目失败
以下代码示例显示如何为接收来自 Kinesis 流的事件的 Lambda 函数实现部分批处理响应。该函数在响应中报告批处理项目失败，并指示 Lambda 稍后重试这些消息。
------
#### [ .NET ]
**适用于 .NET 的 SDK**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-kinesis-to-lambda-with-batch-item-handling)存储库中查找完整示例，并了解如何进行设置和运行。
报告使用 .NET 进行 Lambda Kinesis 批处理项目失败。
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
using System.Text;
using System.Text.Json.Serialization;
using Amazon.Lambda.Core;
using Amazon.Lambda.KinesisEvents;
using AWS.Lambda.Powertools.Logging;
// 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 KinesisIntegration;
public class Function
{
// Powertools Logger requires an environment variables against your function
// POWERTOOLS_SERVICE_NAME
[Logging(LogEvent = true)]
public async Task FunctionHandler(KinesisEvent evnt, ILambdaContext context)
{
if (evnt.Records.Count == 0)
{
Logger.LogInformation("Empty Kinesis Event received");
return new StreamsEventResponse();
}
foreach (var record in evnt.Records)
{
try
{
Logger.LogInformation($"Processed Event with EventId: {record.EventId}");
string data = await GetRecordDataAsync(record.Kinesis, context);
Logger.LogInformation($"Data: {data}");
// TODO: Do interesting work based on the new data
}
catch (Exception ex)
{
Logger.LogError($"An error occurred {ex.Message}");
/* Since we are working with streams, we can return the failed item immediately.
Lambda will immediately begin to retry processing from this failed item onwards. */
return new StreamsEventResponse
{
BatchItemFailures = new List
{
new StreamsEventResponse.BatchItemFailure { ItemIdentifier = record.Kinesis.SequenceNumber }
}
};
}
}
Logger.LogInformation($"Successfully processed {evnt.Records.Count} records.");
return new StreamsEventResponse();
}
private async Task GetRecordDataAsync(KinesisEvent.Record record, ILambdaContext context)
{
byte[] bytes = record.Data.ToArray();
string data = Encoding.UTF8.GetString(bytes);
await Task.CompletedTask; //Placeholder for actual async work
return data;
}
}
public class StreamsEventResponse
{
[JsonPropertyName("batchItemFailures")]
public IList BatchItemFailures { get; set; }
public class BatchItemFailure
{
[JsonPropertyName("itemIdentifier")]
public string ItemIdentifier { get; set; }
}
}
```
------
#### [ Go ]
**适用于 Go 的 SDK V2**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-kinesis-to-lambda-with-batch-item-handling)存储库中查找完整示例，并了解如何进行设置和运行。
报告通过 Go 进行 Lambda Kinesis 批处理项目失败。
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
package main
import (
"context"
"fmt"
"github.com/aws/aws-lambda-go/events"
"github.com/aws/aws-lambda-go/lambda"
)
func handler(ctx context.Context, kinesisEvent events.KinesisEvent) (map[string]interface{}, error) {
batchItemFailures := []map[string]interface{}{}
for _, record := range kinesisEvent.Records {
curRecordSequenceNumber := ""
// Process your record
if /* Your record processing condition here */ {
curRecordSequenceNumber = record.Kinesis.SequenceNumber
}
// Add a condition to check if the record processing failed
if curRecordSequenceNumber != "" {
batchItemFailures = append(batchItemFailures, map[string]interface{}{"itemIdentifier": curRecordSequenceNumber})
}
}
kinesisBatchResponse := map[string]interface{}{
"batchItemFailures": batchItemFailures,
}
return kinesisBatchResponse, nil
}
func main() {
lambda.Start(handler)
}
```
------
#### [ Java ]
**适用于 Java 的 SDK 2.x**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-kinesis-to-lambda-with-batch-item-handling)存储库中查找完整示例，并了解如何进行设置和运行。
报告使用 Java 进行 Lambda Kinesis 批处理项目失败。
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
import com.amazonaws.services.lambda.runtime.Context;
import com.amazonaws.services.lambda.runtime.RequestHandler;
import com.amazonaws.services.lambda.runtime.events.KinesisEvent;
import com.amazonaws.services.lambda.runtime.events.StreamsEventResponse;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.List;
public class ProcessKinesisRecords implements RequestHandler {
@Override
public StreamsEventResponse handleRequest(KinesisEvent input, Context context) {
List batchItemFailures = new ArrayList<>();
String curRecordSequenceNumber = "";
for (KinesisEvent.KinesisEventRecord kinesisEventRecord : input.getRecords()) {
try {
//Process your record
KinesisEvent.Record kinesisRecord = kinesisEventRecord.getKinesis();
curRecordSequenceNumber = kinesisRecord.getSequenceNumber();
} catch (Exception e) {
/* Since we are working with streams, we can return the failed item immediately.
Lambda will immediately begin to retry processing from this failed item onwards. */
batchItemFailures.add(new StreamsEventResponse.BatchItemFailure(curRecordSequenceNumber));
return new StreamsEventResponse(batchItemFailures);
}
}
return new StreamsEventResponse(batchItemFailures);
}
}
```
------
#### [ JavaScript ]
**SDK for JavaScript（v3）**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/blob/main/integration-kinesis-to-lambda-with-batch-item-handling)存储库中查找完整示例，并了解如何进行设置和运行。
报告使用 Javascript 进行 Lambda Kinesis 批处理项目失败。
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
exports.handler = async (event, context) => {
for (const record of event.Records) {
try {
console.log(`Processed Kinesis Event - EventID: ${record.eventID}`);
const recordData = await getRecordDataAsync(record.kinesis);
console.log(`Record Data: ${recordData}`);
// TODO: Do interesting work based on the new data
} catch (err) {
console.error(`An error occurred ${err}`);
/* Since we are working with streams, we can return the failed item immediately.
Lambda will immediately begin to retry processing from this failed item onwards. */
return {
batchItemFailures: [{ itemIdentifier: record.kinesis.sequenceNumber }],
};
}
}
console.log(`Successfully processed ${event.Records.length} records.`);
return { batchItemFailures: [] };
};
async function getRecordDataAsync(payload) {
var data = Buffer.from(payload.data, "base64").toString("utf-8");
await Promise.resolve(1); //Placeholder for actual async work
return data;
}
```
报告使用 TypeScript 进行 Lambda Kinesis 批处理项目失败。
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
import {
KinesisStreamEvent,
Context,
KinesisStreamHandler,
KinesisStreamRecordPayload,
KinesisStreamBatchResponse,
} from "aws-lambda";
import { Buffer } from "buffer";
import { Logger } from "@aws-lambda-powertools/logger";
const logger = new Logger({
logLevel: "INFO",
serviceName: "kinesis-stream-handler-sample",
});
export const functionHandler: KinesisStreamHandler = async (
event: KinesisStreamEvent,
context: Context
): Promise => {
for (const record of event.Records) {
try {
logger.info(`Processed Kinesis Event - EventID: ${record.eventID}`);
const recordData = await getRecordDataAsync(record.kinesis);
logger.info(`Record Data: ${recordData}`);
// TODO: Do interesting work based on the new data
} catch (err) {
logger.error(`An error occurred ${err}`);
/* Since we are working with streams, we can return the failed item immediately.
Lambda will immediately begin to retry processing from this failed item onwards. */
return {
batchItemFailures: [{ itemIdentifier: record.kinesis.sequenceNumber }],
};
}
}
logger.info(`Successfully processed ${event.Records.length} records.`);
return { batchItemFailures: [] };
};
async function getRecordDataAsync(
payload: KinesisStreamRecordPayload
): Promise {
var data = Buffer.from(payload.data, "base64").toString("utf-8");
await Promise.resolve(1); //Placeholder for actual async work
return data;
}
```
------
#### [ PHP ]
**适用于 PHP 的 SDK**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-kinesis-to-lambda-with-batch-item-handling)存储库中查找完整示例，并了解如何进行设置和运行。
报告通过 PHP 进行 Lambda Kinesis 批处理项目失败。
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
logger = $logger;
}
/**
* @throws JsonException
* @throws \Bref\Event\InvalidLambdaEvent
*/
public function handle(mixed $event, Context $context): array
{
$kinesisEvent = new KinesisEvent($event);
$this->logger->info("Processing records");
$records = $kinesisEvent->getRecords();
$failedRecords = [];
foreach ($records as $record) {
try {
$data = $record->getData();
$this->logger->info(json_encode($data));
// TODO: Do interesting work based on the new data
} catch (Exception $e) {
$this->logger->error($e->getMessage());
// failed processing the record
$failedRecords[] = $record->getSequenceNumber();
}
}
$totalRecords = count($records);
$this->logger->info("Successfully processed $totalRecords records");
// change format for the response
$failures = array_map(
fn(string $sequenceNumber) => ['itemIdentifier' => $sequenceNumber],
$failedRecords
);
return [
'batchItemFailures' => $failures
];
}
}
$logger = new StderrLogger();
return new Handler($logger);
```
------
#### [ Python ]
**适用于 Python 的 SDK（Boto3）**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-kinesis-to-lambda-with-batch-item-handling)存储库中查找完整示例，并了解如何进行设置和运行。
报告使用 Python 进行 Lambda Kinesis 批处理项目失败。
```
# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
# SPDX-License-Identifier: Apache-2.0
def handler(event, context):
records = event.get("Records")
curRecordSequenceNumber = ""
for record in records:
try:
# Process your record
curRecordSequenceNumber = record["kinesis"]["sequenceNumber"]
except Exception as e:
# Return failed record's sequence number
return {"batchItemFailures":[{"itemIdentifier": curRecordSequenceNumber}]}
return {"batchItemFailures":[]}
```
------
#### [ Ruby ]
**适用于 Ruby 的 SDK**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-kinesis-to-lambda-with-batch-item-handling)存储库中查找完整示例，并了解如何进行设置和运行。
报告通过 Ruby 进行 Lambda Kinesis 批处理项目失败。
```
# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
# SPDX-License-Identifier: Apache-2.0
require 'aws-sdk'
def lambda_handler(event:, context:)
batch_item_failures = []
event['Records'].each do |record|
begin
puts "Processed Kinesis Event - EventID: #{record['eventID']}"
record_data = get_record_data_async(record['kinesis'])
puts "Record Data: #{record_data}"
# TODO: Do interesting work based on the new data
rescue StandardError => err
puts "An error occurred #{err}"
# Since we are working with streams, we can return the failed item immediately.
# Lambda will immediately begin to retry processing from this failed item onwards.
return { batchItemFailures: [{ itemIdentifier: record['kinesis']['sequenceNumber'] }] }
end
end
puts "Successfully processed #{event['Records'].length} records."
{ batchItemFailures: batch_item_failures }
end
def get_record_data_async(payload)
data = Base64.decode64(payload['data']).force_encoding('utf-8')
# Placeholder for actual async work
sleep(1)
data
end
```
------
#### [ Rust ]
**适用于 Rust 的 SDK**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-kinesis-to-lambda-with-batch-item-handling)存储库中查找完整示例，并了解如何进行设置和运行。
报告通过 Rust 进行 Lambda Kinesis 批处理项目失败。
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
use aws_lambda_events::{
event::kinesis::KinesisEvent,
kinesis::KinesisEventRecord,
streams::{KinesisBatchItemFailure, KinesisEventResponse},
};
use lambda_runtime::{run, service_fn, Error, LambdaEvent};
async fn function_handler(event: LambdaEvent) -> Result {
let mut response = KinesisEventResponse {
batch_item_failures: vec![],
};
if event.payload.records.is_empty() {
tracing::info!("No records found. Exiting.");
return Ok(response);
}
for record in &event.payload.records {
tracing::info!(
"EventId: {}",
record.event_id.as_deref().unwrap_or_default()
);
let record_processing_result = process_record(record);
if record_processing_result.is_err() {
response.batch_item_failures.push(KinesisBatchItemFailure {
item_identifier: record.kinesis.sequence_number.clone(),
});
/* Since we are working with streams, we can return the failed item immediately.
Lambda will immediately begin to retry processing from this failed item onwards. */
return Ok(response);
}
}
tracing::info!(
"Successfully processed {} records",
event.payload.records.len()
);
Ok(response)
}
fn process_record(record: &KinesisEventRecord) -> Result<(), Error> {
let record_data = std::str::from_utf8(record.kinesis.data.as_slice());
if let Some(err) = record_data.err() {
tracing::error!("Error: {}", err);
return Err(Error::from(err));
}
let record_data = record_data.unwrap_or_default();
// do something interesting with the data
tracing::info!("Data: {}", record_data);
Ok(())
}
#[tokio::main]
async fn main() -> Result<(), Error> {
tracing_subscriber::fmt()
.with_max_level(tracing::Level::INFO)
// disable printing the name of the module in every log line.
.with_target(false)
// disabling time is handy because CloudWatch will add the ingestion time.
.without_time()
.init();
run(service_fn(function_handler)).await
}
```
------
有关 AWS SDK 开发人员指南和代码示例的完整列表，请参阅 [将 Lambda 与 AWS SDK 配合使用](sdk-general-information-section.md) 本主题还包括有关入门的信息以及有关先前的 SDK 版本的详细信息。
# 通过 DynamoDB 触发器报告 Lambda 函数批处理项目失败
以下代码示例显示如何为接收来自 DynamoDB 流的事件的 Lambda 函数实现部分批处理响应。该函数在响应中报告批处理项目失败，并指示 Lambda 稍后重试这些消息。
------
#### [ .NET ]
**适用于 .NET 的 SDK**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-ddb-to-lambda-with-batch-item-handling)存储库中查找完整示例，并了解如何进行设置和运行。
报告使用 .NET 通过 Lambda 进行 DynamoDB 批处理项目失败。
```
// 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 StreamsEventResponse FunctionHandler(DynamoDBEvent dynamoEvent, ILambdaContext context)
{
context.Logger.LogInformation($"Beginning to process {dynamoEvent.Records.Count} records...");
List batchItemFailures = new List();
StreamsEventResponse streamsEventResponse = new StreamsEventResponse();
foreach (var record in dynamoEvent.Records)
{
try
{
var sequenceNumber = record.Dynamodb.SequenceNumber;
context.Logger.LogInformation(sequenceNumber);
}
catch (Exception ex)
{
context.Logger.LogError(ex.Message);
batchItemFailures.Add(new StreamsEventResponse.BatchItemFailure() { ItemIdentifier = record.Dynamodb.SequenceNumber });
}
}
if (batchItemFailures.Count > 0)
{
streamsEventResponse.BatchItemFailures = batchItemFailures;
}
context.Logger.LogInformation("Stream processing complete.");
return streamsEventResponse;
}
}
```
------
#### [ Go ]
**适用于 Go 的 SDK V2**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-ddb-to-lambda-with-batch-item-handling)存储库中查找完整示例，并了解如何进行设置和运行。
报告使用 Go 通过 Lambda 进行 DynamoDB 批处理项目失败。
```
// 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/events"
"github.com/aws/aws-lambda-go/lambda"
)
type BatchItemFailure struct {
ItemIdentifier string `json:"ItemIdentifier"`
}
type BatchResult struct {
BatchItemFailures []BatchItemFailure `json:"BatchItemFailures"`
}
func HandleRequest(ctx context.Context, event events.DynamoDBEvent) (*BatchResult, error) {
var batchItemFailures []BatchItemFailure
curRecordSequenceNumber := ""
for _, record := range event.Records {
// Process your record
curRecordSequenceNumber = record.Change.SequenceNumber
}
if curRecordSequenceNumber != "" {
batchItemFailures = append(batchItemFailures, BatchItemFailure{ItemIdentifier: curRecordSequenceNumber})
}
batchResult := BatchResult{
BatchItemFailures: batchItemFailures,
}
return &batchResult, nil
}
func main() {
lambda.Start(HandleRequest)
}
```
------
#### [ Java ]
**适用于 Java 的 SDK 2.x**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-ddb-to-lambda-with-batch-item-handling)存储库中查找完整示例，并了解如何进行设置和运行。
报告使用 Java 通过 Lambda 进行 DynamoDB 批处理项目失败。
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
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.StreamsEventResponse;
import com.amazonaws.services.lambda.runtime.events.models.dynamodb.StreamRecord;
import java.util.ArrayList;
import java.util.List;
public class ProcessDynamodbRecords implements RequestHandler {
@Override
public StreamsEventResponse handleRequest(DynamodbEvent input, Context context) {
List batchItemFailures = new ArrayList<>();
String curRecordSequenceNumber = "";
for (DynamodbEvent.DynamodbStreamRecord dynamodbStreamRecord : input.getRecords()) {
try {
//Process your record
StreamRecord dynamodbRecord = dynamodbStreamRecord.getDynamodb();
curRecordSequenceNumber = dynamodbRecord.getSequenceNumber();
} catch (Exception e) {
/* Since we are working with streams, we can return the failed item immediately.
Lambda will immediately begin to retry processing from this failed item onwards. */
batchItemFailures.add(new StreamsEventResponse.BatchItemFailure(curRecordSequenceNumber));
return new StreamsEventResponse(batchItemFailures);
}
}
return new StreamsEventResponse();
}
}
```
------
#### [ JavaScript ]
**SDK for JavaScript（v3）**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-ddb-to-lambda-with-batch-item-handling)存储库中查找完整示例，并了解如何进行设置和运行。
使用 JavaScript 报告 Lambda 的 DynamoDB 批处理项目失败。
```
export const handler = async (event) => {
const records = event.Records;
let curRecordSequenceNumber = "";
for (const record of records) {
try {
// Process your record
curRecordSequenceNumber = record.dynamodb.SequenceNumber;
} catch (e) {
// Return failed record's sequence number
return { batchItemFailures: [{ itemIdentifier: curRecordSequenceNumber }] };
}
}
return { batchItemFailures: [] };
};
```
使用 TypeScript 报告 Lambda 的 DynamoDB 批处理项目失败。
```
import {
DynamoDBBatchResponse,
DynamoDBBatchItemFailure,
DynamoDBStreamEvent,
} from "aws-lambda";
export const handler = async (
event: DynamoDBStreamEvent
): Promise => {
const batchItemFailures: DynamoDBBatchItemFailure[] = [];
let curRecordSequenceNumber;
for (const record of event.Records) {
curRecordSequenceNumber = record.dynamodb?.SequenceNumber;
if (curRecordSequenceNumber) {
batchItemFailures.push({
itemIdentifier: curRecordSequenceNumber,
});
}
}
return { batchItemFailures: batchItemFailures };
};
```
------
#### [ PHP ]
**适用于 PHP 的 SDK**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-ddb-to-lambda-with-batch-item-handling)存储库中查找完整示例，并了解如何进行设置和运行。
报告使用 PHP 通过 Lambda 进行 DynamoDB 批处理项目失败。
```
logger = $logger;
}
/**
* @throws JsonException
* @throws \Bref\Event\InvalidLambdaEvent
*/
public function handle(mixed $event, Context $context): array
{
$dynamoDbEvent = new DynamoDbEvent($event);
$this->logger->info("Processing records");
$records = $dynamoDbEvent->getRecords();
$failedRecords = [];
foreach ($records as $record) {
try {
$data = $record->getData();
$this->logger->info(json_encode($data));
// TODO: Do interesting work based on the new data
} catch (Exception $e) {
$this->logger->error($e->getMessage());
// failed processing the record
$failedRecords[] = $record->getSequenceNumber();
}
}
$totalRecords = count($records);
$this->logger->info("Successfully processed $totalRecords records");
// change format for the response
$failures = array_map(
fn(string $sequenceNumber) => ['itemIdentifier' => $sequenceNumber],
$failedRecords
);
return [
'batchItemFailures' => $failures
];
}
}
$logger = new StderrLogger();
return new Handler($logger);
```
------
#### [ Python ]
**适用于 Python 的 SDK（Boto3）**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-ddb-to-lambda-with-batch-item-handling)存储库中查找完整示例，并了解如何进行设置和运行。
报告使用 Python 通过 Lambda 进行 DynamoDB 批处理项目失败。
```
# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
# SPDX-License-Identifier: Apache-2.0
def handler(event, context):
records = event.get("Records")
curRecordSequenceNumber = ""
for record in records:
try:
# Process your record
curRecordSequenceNumber = record["dynamodb"]["SequenceNumber"]
except Exception as e:
# Return failed record's sequence number
return {"batchItemFailures":[{"itemIdentifier": curRecordSequenceNumber}]}
return {"batchItemFailures":[]}
```
------
#### [ Ruby ]
**适用于 Ruby 的 SDK**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-ddb-to-lambda-with-batch-item-handling)存储库中查找完整示例，并了解如何进行设置和运行。
报告使用 Ruby 通过 Lambda 进行 DynamoDB 批处理项目失败。
```
def lambda_handler(event:, context:)
records = event["Records"]
cur_record_sequence_number = ""
records.each do |record|
begin
# Process your record
cur_record_sequence_number = record["dynamodb"]["SequenceNumber"]
rescue StandardError => e
# Return failed record's sequence number
return {"batchItemFailures" => [{"itemIdentifier" => cur_record_sequence_number}]}
end
end
{"batchItemFailures" => []}
end
```
------
#### [ Rust ]
**适用于 Rust 的 SDK**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-ddb-to-lambda-with-batch-item-handling)存储库中查找完整示例，并了解如何进行设置和运行。
报告使用 Rust 通过 Lambda 进行 DynamoDB 批处理项目失败。
```
use aws_lambda_events::{
event::dynamodb::{Event, EventRecord, StreamRecord},
streams::{DynamoDbBatchItemFailure, DynamoDbEventResponse},
};
use lambda_runtime::{run, service_fn, Error, LambdaEvent};
/// Process the stream record
fn process_record(record: &EventRecord) -> Result<(), Error> {
let stream_record: &StreamRecord = &record.change;
// process your stream record here...
tracing::info!("Data: {:?}", stream_record);
Ok(())
}
/// Main Lambda handler here...
async fn function_handler(event: LambdaEvent) -> Result {
let mut response = DynamoDbEventResponse {
batch_item_failures: vec![],
};
let records = &event.payload.records;
if records.is_empty() {
tracing::info!("No records found. Exiting.");
return Ok(response);
}
for record in records {
tracing::info!("EventId: {}", record.event_id);
// Couldn't find a sequence number
if record.change.sequence_number.is_none() {
response.batch_item_failures.push(DynamoDbBatchItemFailure {
item_identifier: Some("".to_string()),
});
return Ok(response);
}
// Process your record here...
if process_record(record).is_err() {
response.batch_item_failures.push(DynamoDbBatchItemFailure {
item_identifier: record.change.sequence_number.clone(),
});
/* Since we are working with streams, we can return the failed item immediately.
Lambda will immediately begin to retry processing from this failed item onwards. */
return Ok(response);
}
}
tracing::info!("Successfully processed {} record(s)", records.len());
Ok(response)
}
#[tokio::main]
async fn main() -> Result<(), Error> {
tracing_subscriber::fmt()
.with_max_level(tracing::Level::INFO)
// disable printing the name of the module in every log line.
.with_target(false)
// disabling time is handy because CloudWatch will add the ingestion time.
.without_time()
.init();
run(service_fn(function_handler)).await
}
```
------
有关 AWS SDK 开发人员指南和代码示例的完整列表，请参阅 [将 Lambda 与 AWS SDK 配合使用](sdk-general-information-section.md) 本主题还包括有关入门的信息以及有关先前的 SDK 版本的详细信息。
# 报告使用 Amazon SQS 触发器进行 Lambda 函数批处理项目失败
以下代码示例显示如何为接收来自 SQS 队列的事件的 Lambda 函数实现部分批处理响应。该函数在响应中报告批处理项目失败，并指示 Lambda 稍后重试这些消息。
------
#### [ .NET ]
**适用于 .NET 的 SDK**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/lambda-function-sqs-report-batch-item-failures)存储库中查找完整示例，并了解如何进行设置和运行。
报告使用 .NET 进行 Lambda SQS 批处理项目失败。
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
using Amazon.Lambda.Core;
using Amazon.Lambda.SQSEvents;
// 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 sqsSample;
public class Function
{
public async Task FunctionHandler(SQSEvent evnt, ILambdaContext context)
{
List batchItemFailures = new List();
foreach(var message in evnt.Records)
{
try
{
//process your message
await ProcessMessageAsync(message, context);
}
catch (System.Exception)
{
//Add failed message identifier to the batchItemFailures list
batchItemFailures.Add(new SQSBatchResponse.BatchItemFailure{ItemIdentifier=message.MessageId});
}
}
return new SQSBatchResponse(batchItemFailures);
}
private async Task ProcessMessageAsync(SQSEvent.SQSMessage message, ILambdaContext context)
{
if (String.IsNullOrEmpty(message.Body))
{
throw new Exception("No Body in SQS Message.");
}
context.Logger.LogInformation($"Processed message {message.Body}");
// TODO: Do interesting work based on the new message
await Task.CompletedTask;
}
}
```
------
#### [ Go ]
**适用于 Go 的 SDK V2**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/lambda-function-sqs-report-batch-item-failures)存储库中查找完整示例，并了解如何进行设置和运行。
报告使用 Go 进行 Lambda SQS 批处理项目失败。
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
package main
import (
"context"
"fmt"
"github.com/aws/aws-lambda-go/events"
"github.com/aws/aws-lambda-go/lambda"
)
func handler(ctx context.Context, sqsEvent events.SQSEvent) (map[string]interface{}, error) {
batchItemFailures := []map[string]interface{}{}
for _, message := range sqsEvent.Records {
if len(message.Body) > 0 {
// Your message processing condition here
fmt.Printf("Successfully processed message: %s\n", message.Body)
} else {
// Message processing failed
fmt.Printf("Failed to process message %s\n", message.MessageId)
batchItemFailures = append(batchItemFailures, map[string]interface{}{"itemIdentifier": message.MessageId})
}
}
sqsBatchResponse := map[string]interface{}{
"batchItemFailures": batchItemFailures,
}
return sqsBatchResponse, nil
}
func main() {
lambda.Start(handler)
}
```
------
#### [ Java ]
**适用于 Java 的 SDK 2.x**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/lambda-function-sqs-report-batch-item-failures)存储库中查找完整示例，并了解如何进行设置和运行。
报告使用 Java 进行 Lambda SQS 批处理项目失败。
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
import com.amazonaws.services.lambda.runtime.Context;
import com.amazonaws.services.lambda.runtime.RequestHandler;
import com.amazonaws.services.lambda.runtime.events.SQSEvent;
import com.amazonaws.services.lambda.runtime.events.SQSBatchResponse;
import java.util.ArrayList;
import java.util.List;
public class ProcessSQSMessageBatch implements RequestHandler {
@Override
public SQSBatchResponse handleRequest(SQSEvent sqsEvent, Context context) {
List batchItemFailures = new ArrayList();
for (SQSEvent.SQSMessage message : sqsEvent.getRecords()) {
try {
//process your message
} catch (Exception e) {
//Add failed message identifier to the batchItemFailures list
batchItemFailures.add(new SQSBatchResponse.BatchItemFailure(message.getMessageId()));
}
}
return new SQSBatchResponse(batchItemFailures);
}
}
```
------
#### [ JavaScript ]
**SDK for JavaScript（v3）**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/lambda-function-sqs-report-batch-item-failures)存储库中查找完整示例，并了解如何进行设置和运行。
报告使用 JavaScript 进行 Lambda SQS 批处理项目失败。
```
// Node.js 20.x Lambda runtime, AWS SDK for Javascript V3
export const handler = async (event, context) => {
const batchItemFailures = [];
for (const record of event.Records) {
try {
await processMessageAsync(record, context);
} catch (error) {
batchItemFailures.push({ itemIdentifier: record.messageId });
}
}
return { batchItemFailures };
};
async function processMessageAsync(record, context) {
if (record.body && record.body.includes("error")) {
throw new Error("There is an error in the SQS Message.");
}
console.log(`Processed message: ${record.body}`);
}
```
报告使用 TypeScript 进行 Lambda SQS 批处理项目失败。
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
import { SQSEvent, SQSBatchResponse, Context, SQSBatchItemFailure, SQSRecord } from 'aws-lambda';
export const handler = async (event: SQSEvent, context: Context): Promise => {
const batchItemFailures: SQSBatchItemFailure[] = [];
for (const record of event.Records) {
try {
await processMessageAsync(record);
} catch (error) {
batchItemFailures.push({ itemIdentifier: record.messageId });
}
}
return {batchItemFailures: batchItemFailures};
};
async function processMessageAsync(record: SQSRecord): Promise {
if (record.body && record.body.includes("error")) {
throw new Error('There is an error in the SQS Message.');
}
console.log(`Processed message ${record.body}`);
}
```
------
#### [ PHP ]
**适用于 PHP 的 SDK**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/lambda-function-sqs-report-batch-item-failures)存储库中查找完整示例，并了解如何进行设置和运行。
报告使用 PHP 进行 Lambda SQS 批处理项目失败。
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
logger = $logger;
}
/**
* @throws JsonException
* @throws \Bref\Event\InvalidLambdaEvent
*/
public function handleSqs(SqsEvent $event, Context $context): void
{
$this->logger->info("Processing SQS records");
$records = $event->getRecords();
foreach ($records as $record) {
try {
// Assuming the SQS message is in JSON format
$message = json_decode($record->getBody(), true);
$this->logger->info(json_encode($message));
// TODO: Implement your custom processing logic here
} catch (Exception $e) {
$this->logger->error($e->getMessage());
// failed processing the record
$this->markAsFailed($record);
}
}
$totalRecords = count($records);
$this->logger->info("Successfully processed $totalRecords SQS records");
}
}
$logger = new StderrLogger();
return new Handler($logger);
```
------
#### [ Python ]
**适用于 Python 的 SDK（Boto3）**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/lambda-function-sqs-report-batch-item-failures)存储库中查找完整示例，并了解如何进行设置和运行。
报告使用 Python 进行 Lambda SQS 批处理项目失败。
```
# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
# SPDX-License-Identifier: Apache-2.0
def lambda_handler(event, context):
if event:
batch_item_failures = []
sqs_batch_response = {}
for record in event["Records"]:
try:
print(f"Processed message: {record['body']}")
except Exception as e:
batch_item_failures.append({"itemIdentifier": record['messageId']})
sqs_batch_response["batchItemFailures"] = batch_item_failures
return sqs_batch_response
```
------
#### [ Ruby ]
**适用于 Ruby 的 SDK**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/integration-sqs-to-lambda-with-batch-item-handling)存储库中查找完整示例，并了解如何进行设置和运行。
报告使用 Ruby 进行 Lambda SQS 批处理项目失败。
```
# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
# SPDX-License-Identifier: Apache-2.0
require 'json'
def lambda_handler(event:, context:)
if event
batch_item_failures = []
sqs_batch_response = {}
event["Records"].each do |record|
begin
# process message
rescue StandardError => e
batch_item_failures << {"itemIdentifier" => record['messageId']}
end
end
sqs_batch_response["batchItemFailures"] = batch_item_failures
return sqs_batch_response
end
end
```
------
#### [ Rust ]
**适用于 Rust 的 SDK**
查看 GitHub，了解更多信息。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/tree/main/lambda-function-sqs-report-batch-item-failures)存储库中查找完整示例，并了解如何进行设置和运行。
报告使用 Rust 进行 Lambda SQS 批处理项目失败。
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
use aws_lambda_events::{
event::sqs::{SqsBatchResponse, SqsEvent},
sqs::{BatchItemFailure, SqsMessage},
};
use lambda_runtime::{run, service_fn, Error, LambdaEvent};
async fn process_record(_: &SqsMessage) -> Result<(), Error> {
Err(Error::from("Error processing message"))
}
async fn function_handler(event: LambdaEvent) -> Result {
let mut batch_item_failures = Vec::new();
for record in event.payload.records {
match process_record(&record).await {
Ok(_) => (),
Err(_) => batch_item_failures.push(BatchItemFailure {
item_identifier: record.message_id.unwrap(),
}),
}
}
Ok(SqsBatchResponse {
batch_item_failures,
})
}
#[tokio::main]
async fn main() -> Result<(), Error> {
run(service_fn(function_handler)).await
}
```
------
有关 AWS SDK 开发人员指南和代码示例的完整列表，请参阅 [将 Lambda 与 AWS SDK 配合使用](sdk-general-information-section.md) 本主题还包括有关入门的信息以及有关先前的 SDK 版本的详细信息。