文件 AWS 開發套件範例 GitHub 儲存庫中有更多可用的 AWS SDK 範例
本文為英文版的機器翻譯版本,如內容有任何歧義或不一致之處,概以英文版為準。
使用適用於 Ruby 的 SDK 的 Kinesis 範例
下列程式碼範例示範如何使用 AWS SDK for Ruby 搭配 Kinesis 來執行動作和實作常見案例。
每個範例都包含完整原始程式碼的連結,您可以在其中找到如何在內容中設定和執行程式碼的指示。
主題
無伺服器範例
下列程式碼範例示範如何實作 Lambda 函數,該函數接收從 Kinesis 串流接收記錄所觸發的事件。此函數會擷取 Kinesis 承載、從 Base64 解碼,並記錄記錄內容。
- SDK for Ruby
-
注意
GitHub 上提供更多範例。尋找完整範例,並了解如何在無伺服器範例
儲存庫中設定和執行。 使用 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:) 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
下列程式碼範例示範如何為從 Kinesis 串流接收事件的 Lambda 函數實作部分批次回應。此函數會在回應中報告批次項目失敗,指示 Lambda 稍後重試這些訊息。
- SDK for Ruby
-
注意
GitHub 上提供更多範例。尋找完整範例,並了解如何在無伺服器範例
儲存庫中設定和執行。 使用 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