[适用于 JavaScript 的 AWS SDK V3 API 参考指南](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/)详细描述了 适用于 JavaScript 的 AWS SDK 版本 3 (V3) 的所有 API 操作。 本文属于机器翻译版本。若本译文内容与英语原文存在差异，则一律以英文原文为准。 # 使用适用于 JavaScript (v3) 的 SDK 的 Kinesis 示例 以下代码示例向您展示了如何使用带有 Kinesis 的 适用于 JavaScript 的 AWS SDK (v3) 来执行操作和实现常见场景。 *操作*是大型程序的代码摘录，必须在上下文中运行。您可以通过操作了解如何调用单个服务函数，还可以通过函数相关场景的上下文查看操作。 每个示例都包含一个指向完整源代码的链接，您可以从中找到有关如何在上下文中设置和运行代码的说明。 **Topics** + [操作](#actions) + [无服务器示例](#serverless_examples) ## 操作 ### `PutRecords` 以下代码示例演示了如何使用 `PutRecords`。 **适用于 JavaScript (v3) 的软件开发工具包** 还有更多相关信息 GitHub。在 [AWS 代码示例存储库](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/javascriptv3/example_code/kinesis#code-examples)中查找完整示例，了解如何进行设置和运行。 ``` import { PutRecordsCommand, KinesisClient } from "@aws-sdk/client-kinesis"; /** * Put multiple records into a Kinesis stream. * @param {{ streamArn: string }} config */ export const main = async ({ streamArn }) => { const client = new KinesisClient({}); try { await client.send( new PutRecordsCommand({ StreamARN: streamArn, Records: [ { Data: new Uint8Array(), /** * Determines which shard in the stream the data record is assigned to. * Partition keys are Unicode strings with a maximum length limit of 256 * characters for each key. Amazon Kinesis Data Streams uses the partition * key as input to a hash function that maps the partition key and * associated data to a specific shard. */ PartitionKey: "TEST_KEY", }, { Data: new Uint8Array(), PartitionKey: "TEST_KEY", }, ], }), ); } catch (caught) { if (caught instanceof Error) { // } else { throw caught; } } }; // Call function if run directly. import { fileURLToPath } from "node:url"; import { parseArgs } from "node:util"; if (process.argv[1] === fileURLToPath(import.meta.url)) { const options = { streamArn: { type: "string", description: "The ARN of the stream.", }, }; const { values } = parseArgs({ options }); main(values); } ``` + 有关 API 的详细信息，请参阅 *适用于 JavaScript 的 AWS SDK API 参考[PutRecords](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/client/kinesis/command/PutRecordsCommand)*中的。 ## 无服务器示例 ### 通过 Kinesis 触发器调用 Lambda 函数 以下代码示例展示了如何实现一个 Lambda 函数，该函数接收因接收来自 Kinesis 流的记录而触发的事件。该函数检索 Kinesis 有效负载，将 Base64 解码，并记录下记录内容。 **适用于 JavaScript (v3) 的软件开发工具包** 还有更多相关信息 GitHub。在[无服务器示例](https://github.com/aws-samples/serverless-snippets/blob/main/integration-kinesis-to-lambda)存储库中查找完整示例，并了解如何进行设置和运行。 使用 Lambda 消耗 Kinesis 事件。 JavaScript ``` // 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; } ``` 使用 Lambda 消耗 Kinesis 事件。 TypeScript ``` // 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; } ``` ### 通过 Kinesis 触发器报告 Lambda 函数批处理项目失败 以下代码示例展示了如何为接收来自 Kinesis 流的事件的 Lambda 函数实现部分批处理响应。该函数在响应中报告批处理项目失败，并指示 Lambda 稍后重试这些消息。 **适用于 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; } ``` 使用 Lambda 报告 Kinesis 批处理项目失败。 TypeScript ``` // 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; } ```