本文属于机器翻译版本。若本译文内容与英语原文存在差异，则一律以英文原文为准。 # 使用 Firehose 的基本示例 AWS SDKs 以下代码示例展示了如何使用 Amazon Data Firehose 的基础知识。 AWS SDKs **Contents** + [操作](service_code_examples_actions.md) + [`PutRecord`](example_firehose_PutRecord_section.md) + [`PutRecordBatch`](example_firehose_PutRecordBatch_section.md) # 使用 Firehose 的操作 AWS SDKs 以下代码示例演示了如何使用执行各个 Firehose 操作。 AWS SDKs每个示例都包含一个指向的链接 GitHub，您可以在其中找到有关设置和运行代码的说明。 这些代码节选调用了 Firehose API，是必须在上下文中运行的大型程序的代码节选。您可以在[使用 Firehose 的场景 AWS SDKs](service_code_examples_scenarios.md)中结合上下文查看操作。 以下示例仅包括最常用的操作。有关完整列表，请参阅 [Amazon Data Firehose API 参考](https://docs.aws.amazon.com/firehose/latest/APIReference/Welcome.html)。 **Topics** + [`PutRecord`](example_firehose_PutRecord_section.md) + [`PutRecordBatch`](example_firehose_PutRecordBatch_section.md) # `PutRecord`与 AWS SDK 或 CLI 配合使用 以下代码示例演示如何使用 `PutRecord`。 操作示例是大型程序的代码摘录，必须在上下文中运行。在以下代码示例中，您可以查看此操作的上下文： + [将记录放入 Firehose](example_firehose_Scenario_PutRecords_section.md) ------ #### [ CLI ] **AWS CLI** **要将记录写入流** 以下 `put-record` 示例将数据写入流中。数据以 Base64 格式编码。 ``` aws firehose put-record \ --delivery-stream-name my-stream \ --record '{"Data":"SGVsbG8gd29ybGQ="}' ``` 输出： ``` { "RecordId": "RjB5K/nnoGFHqwTsZlNd/TTqvjE8V5dsyXZTQn2JXrdpMTOwssyEb6nfC8fwf1whhwnItt4mvrn+gsqeK5jB7QjuLg283+Ps4Sz/j1Xujv31iDhnPdaLw4BOyM9Amv7PcCuB2079RuM0NhoakbyUymlwY8yt20G8X2420wu1jlFafhci4erAt7QhDEvpwuK8N1uOQ1EuaKZWxQHDzcG6tk1E49IPeD9k", "Encrypted": false } ``` 有关更多信息，请参阅《Amazon Kinesis Data Firehose 开发人员指南》**中的[将数据发送到 Amazon Kinesis Data Firehose 传输流](https://docs.aws.amazon.com/firehose/latest/dev/basic-write.html)。 + 有关 API 的详细信息，请参阅*AWS CLI 命令参考[PutRecord](https://awscli.amazonaws.com/v2/documentation/api/latest/reference/firehose/put-record.html)*中的。 ------ #### [ Java ] **适用于 Java 的 SDK 2.x** 还有更多相关信息 GitHub。在 [AWS 代码示例存储库](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/javav2/example_code/firehose#code-examples)中查找完整示例，了解如何进行设置和运行。 ``` /** * Puts a record to the specified Amazon Kinesis Data Firehose delivery stream. * * @param record The record to be put to the delivery stream. The record must be a {@link Map} of String keys and Object values. * @param deliveryStreamName The name of the Amazon Kinesis Data Firehose delivery stream to which the record should be put. * @throws IllegalArgumentException if the input record or delivery stream name is null or empty. * @throws RuntimeException if there is an error putting the record to the delivery stream. */ public static void putRecord(Map record, String deliveryStreamName) { if (record == null || deliveryStreamName == null || deliveryStreamName.isEmpty()) { throw new IllegalArgumentException("Invalid input: record or delivery stream name cannot be null/empty"); } try { String jsonRecord = new ObjectMapper().writeValueAsString(record); Record firehoseRecord = Record.builder() .data(SdkBytes.fromByteArray(jsonRecord.getBytes(StandardCharsets.UTF_8))) .build(); PutRecordRequest putRecordRequest = PutRecordRequest.builder() .deliveryStreamName(deliveryStreamName) .record(firehoseRecord) .build(); getFirehoseClient().putRecord(putRecordRequest); System.out.println("Record sent: " + jsonRecord); } catch (Exception e) { throw new RuntimeException("Failed to put record: " + e.getMessage(), e); } } ``` + 有关 API 的详细信息，请参阅 *AWS SDK for Java 2.x API 参考[PutRecord](https://docs.aws.amazon.com/goto/SdkForJavaV2/firehose-2015-08-04/PutRecord)*中的。 ------ #### [ Python ] **适用于 Python 的 SDK（Boto3）** 还有更多相关信息 GitHub。在 [AWS 代码示例存储库](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/firehose#code-examples)中查找完整示例，了解如何进行设置和运行。 ``` class FirehoseClient: """ AWS Firehose client to send records and monitor metrics. Attributes: config (object): Configuration object with delivery stream name and region. delivery_stream_name (str): Name of the Firehose delivery stream. region (str): AWS region for Firehose and CloudWatch clients. firehose (boto3.client): Boto3 Firehose client. cloudwatch (boto3.client): Boto3 CloudWatch client. """ def __init__(self, config): """ Initialize the FirehoseClient. Args: config (object): Configuration object with delivery stream name and region. """ self.config = config self.delivery_stream_name = config.delivery_stream_name self.region = config.region self.firehose = boto3.client("firehose", region_name=self.region) self.cloudwatch = boto3.client("cloudwatch", region_name=self.region) @backoff.on_exception( backoff.expo, Exception, max_tries=5, jitter=backoff.full_jitter ) def put_record(self, record: dict): """ Put individual records to Firehose with backoff and retry. Args: record (dict): The data record to be sent to Firehose. This method attempts to send an individual record to the Firehose delivery stream. It retries with exponential backoff in case of exceptions. """ try: entry = self._create_record_entry(record) response = self.firehose.put_record( DeliveryStreamName=self.delivery_stream_name, Record=entry ) self._log_response(response, entry) except Exception: logger.info(f"Fail record: {record}.") raise ``` + 有关 API 的详细信息，请参阅适用[PutRecord](https://docs.aws.amazon.com/goto/boto3/firehose-2015-08-04/PutRecord)于 *Python 的AWS SDK (Boto3) API 参考*。 ------ #### [ SAP ABAP ] **适用于 SAP ABAP 的 SDK** 还有更多相关信息 GitHub。在 [AWS 代码示例存储库](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/sap-abap/services/frh#code-examples)中查找完整示例，了解如何进行设置和运行。 ``` TRY. DATA(lo_record) = NEW /aws1/cl_frhrecord( iv_data = iv_data ). DATA(lo_result) = lo_frh->putrecord( iv_deliverystreamname = iv_deliv_stream_name io_record = lo_record ). MESSAGE 'Record sent to Firehose delivery stream.' TYPE 'I'. CATCH /aws1/cx_frhresourcenotfoundex. MESSAGE 'Delivery stream not found.' TYPE 'E'. CATCH /aws1/cx_frhinvalidargumentex. MESSAGE 'Invalid argument provided.' TYPE 'E'. CATCH /aws1/cx_frhserviceunavailex. MESSAGE 'Service temporarily unavailable.' TYPE 'E'. ENDTRY. ``` + 有关 API 的详细信息，请参阅适用[PutRecord](https://docs.aws.amazon.com/sdk-for-sap-abap/v1/api/latest/index.html)于 S *AP 的AWS SDK ABAP API 参考*。 ------ 有关 S AWS DK 开发者指南和代码示例的完整列表，请参阅[将 Firehose 与 SDK 配合使用 AWS](sdk-general-information-section.md)。本主题还包括有关入门的信息以及有关先前的 SDK 版本的详细信息。 # `PutRecordBatch`与 AWS SDK 或 CLI 配合使用 以下代码示例演示如何使用 `PutRecordBatch`。 操作示例是大型程序的代码摘录，必须在上下文中运行。在以下代码示例中，您可以查看此操作的上下文： + [将记录放入 Firehose](example_firehose_Scenario_PutRecords_section.md) ------ #### [ CLI ] **AWS CLI** **将多条记录写入流中** 以下 `put-record-batch` 示例将三条记录写入流中。数据以 Base64 格式编码。 ``` aws firehose put-record-batch \ --delivery-stream-name my-stream \ --records file://records.json ``` `myfile.json` 的内容： ``` [ {"Data": "Rmlyc3QgdGhpbmc="}, {"Data": "U2Vjb25kIHRoaW5n"}, {"Data": "VGhpcmQgdGhpbmc="} ] ``` 输出： ``` { "FailedPutCount": 0, "Encrypted": false, "RequestResponses": [ { "RecordId": "9D2OJ6t2EqCTZTXwGzeSv/EVHxRoRCw89xd+o3+sXg8DhYOaWKPSmZy/CGlRVEys1u1xbeKh6VofEYKkoeiDrcjrxhQp9iF7sUW7pujiMEQ5LzlrzCkGosxQn+3boDnURDEaD42V7GiixpOyLJkYZcae1i7HzlCEoy9LJhMr8EjDSi4Om/9Vc2uhwwuAtGE0XKpxJ2WD7ZRWtAnYlKAnvgSPRgg7zOWL" }, { "RecordId": "jFirejqxCLlK5xjH/UNmlMVcjktEN76I7916X9PaZ+PVaOSXDfU1WGOqEZhxq2js7xcZ552eoeDxsuTU1MSq9nZTbVfb6cQTIXnm/GsuF37Uhg67GKmR5z90l6XKJ+/+pDloFv7Hh9a3oUS6wYm3DcNRLTHHAimANp1PhkQvWpvLRfzbuCUkBphR2QVzhP9OiHLbzGwy8/DfH8sqWEUYASNJKS8GXP5s" }, { "RecordId": "oy0amQ40o5Y2YV4vxzufdcMOOw6n3EPr3tpPJGoYVNKH4APPVqNcbUgefo1stEFRg4hTLrf2k6eliHu/9+YJ5R3iiedHkdsfkIqX0XTySSutvgFYTjNY1TSrK0pM2sWxpjqqnk3+2UX1MV5z88xGro3cQm/DTBt3qBlmTj7Xq8SKVbO1S7YvMTpWkMKA86f8JfmT8BMKoMb4XZS/sOkQLe+qh0sYKXWl" } ] } ``` 有关更多信息，请参阅《Amazon Kinesis Data Firehose 开发人员指南》**中的[将数据发送到 Amazon Kinesis Data Firehose 传输流](https://docs.aws.amazon.com/firehose/latest/dev/basic-write.html)。 + 有关 API 的详细信息，请参阅*AWS CLI 命令参考[PutRecordBatch](https://awscli.amazonaws.com/v2/documentation/api/latest/reference/firehose/put-record-batch.html)*中的。 ------ #### [ Java ] **适用于 Java 的 SDK 2.x** 还有更多相关信息 GitHub。在 [AWS 代码示例存储库](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/javav2/example_code/firehose#code-examples)中查找完整示例，了解如何进行设置和运行。 ``` /** * Puts a batch of records to an Amazon Kinesis Data Firehose delivery stream. * * @param records a list of maps representing the records to be sent * @param batchSize the maximum number of records to include in each batch * @param deliveryStreamName the name of the Kinesis Data Firehose delivery stream * @throws IllegalArgumentException if the input parameters are invalid (null or empty) * @throws RuntimeException if there is an error putting the record batch */ public static void putRecordBatch(List> records, int batchSize, String deliveryStreamName) { if (records == null || records.isEmpty() || deliveryStreamName == null || deliveryStreamName.isEmpty()) { throw new IllegalArgumentException("Invalid input: records or delivery stream name cannot be null/empty"); } ObjectMapper objectMapper = new ObjectMapper(); try { for (int i = 0; i < records.size(); i += batchSize) { List> batch = records.subList(i, Math.min(i + batchSize, records.size())); List batchRecords = batch.stream().map(record -> { try { String jsonRecord = objectMapper.writeValueAsString(record); return Record.builder() .data(SdkBytes.fromByteArray(jsonRecord.getBytes(StandardCharsets.UTF_8))) .build(); } catch (Exception e) { throw new RuntimeException("Error creating Firehose record", e); } }).collect(Collectors.toList()); PutRecordBatchRequest request = PutRecordBatchRequest.builder() .deliveryStreamName(deliveryStreamName) .records(batchRecords) .build(); PutRecordBatchResponse response = getFirehoseClient().putRecordBatch(request); if (response.failedPutCount() > 0) { response.requestResponses().stream() .filter(r -> r.errorCode() != null) .forEach(r -> System.err.println("Failed record: " + r.errorMessage())); } System.out.println("Batch sent with size: " + batchRecords.size()); } } catch (Exception e) { throw new RuntimeException("Failed to put record batch: " + e.getMessage(), e); } } ``` + 有关 API 的详细信息，请参阅 *AWS SDK for Java 2.x API 参考[PutRecordBatch](https://docs.aws.amazon.com/goto/SdkForJavaV2/firehose-2015-08-04/PutRecordBatch)*中的。 ------ #### [ Python ] **适用于 Python 的 SDK（Boto3）** 还有更多相关信息 GitHub。在 [AWS 代码示例存储库](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/firehose#code-examples)中查找完整示例，了解如何进行设置和运行。 ``` class FirehoseClient: """ AWS Firehose client to send records and monitor metrics. Attributes: config (object): Configuration object with delivery stream name and region. delivery_stream_name (str): Name of the Firehose delivery stream. region (str): AWS region for Firehose and CloudWatch clients. firehose (boto3.client): Boto3 Firehose client. cloudwatch (boto3.client): Boto3 CloudWatch client. """ def __init__(self, config): """ Initialize the FirehoseClient. Args: config (object): Configuration object with delivery stream name and region. """ self.config = config self.delivery_stream_name = config.delivery_stream_name self.region = config.region self.firehose = boto3.client("firehose", region_name=self.region) self.cloudwatch = boto3.client("cloudwatch", region_name=self.region) @backoff.on_exception( backoff.expo, Exception, max_tries=5, jitter=backoff.full_jitter ) def put_record_batch(self, data: list, batch_size: int = 500): """ Put records in batches to Firehose with backoff and retry. Args: data (list): List of data records to be sent to Firehose. batch_size (int): Number of records to send in each batch. Default is 500. This method attempts to send records in batches to the Firehose delivery stream. It retries with exponential backoff in case of exceptions. """ for i in range(0, len(data), batch_size): batch = data[i : i + batch_size] record_dicts = [{"Data": json.dumps(record)} for record in batch] try: response = self.firehose.put_record_batch( DeliveryStreamName=self.delivery_stream_name, Records=record_dicts ) self._log_batch_response(response, len(batch)) except Exception as e: logger.info(f"Failed to send batch of {len(batch)} records. Error: {e}") ``` + 有关 API 的详细信息，请参阅适用[PutRecordBatch](https://docs.aws.amazon.com/goto/boto3/firehose-2015-08-04/PutRecordBatch)于 *Python 的AWS SDK (Boto3) API 参考*。 ------ #### [ Rust ] **适用于 Rust 的 SDK** 还有更多相关信息 GitHub。在 [AWS 代码示例存储库](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/rustv1/examples/firehose#code-examples)中查找完整示例，了解如何进行设置和运行。 ``` async fn put_record_batch( client: &Client, stream: &str, data: Vec, ) -> Result> { client .put_record_batch() .delivery_stream_name(stream) .set_records(Some(data)) .send() .await } ``` + 有关 API 的详细信息，请参阅适用[PutRecordBatch](https://docs.rs/aws-sdk-firehose/latest/aws_sdk_firehose/client/struct.Client.html#method.put_record_batch)于 *Rust 的AWS SDK API 参考*。 ------ #### [ SAP ABAP ] **适用于 SAP ABAP 的 SDK** 还有更多相关信息 GitHub。在 [AWS 代码示例存储库](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/sap-abap/services/frh#code-examples)中查找完整示例，了解如何进行设置和运行。 ``` TRY. DATA(lo_result) = lo_frh->putrecordbatch( iv_deliverystreamname = iv_deliv_stream_name it_records = it_records ). DATA(lv_failed_count) = lo_result->get_failedputcount( ). IF lv_failed_count > 0. MESSAGE |{ lv_failed_count } records failed to send.| TYPE 'I'. ELSE. MESSAGE 'All records sent successfully to Firehose delivery stream.' TYPE 'I'. ENDIF. CATCH /aws1/cx_frhresourcenotfoundex. MESSAGE 'Delivery stream not found.' TYPE 'E'. CATCH /aws1/cx_frhinvalidargumentex. MESSAGE 'Invalid argument provided.' TYPE 'E'. CATCH /aws1/cx_frhserviceunavailex. MESSAGE 'Service temporarily unavailable.' TYPE 'E'. ENDTRY. ``` + 有关 API 的详细信息，请参阅适用[PutRecordBatch](https://docs.aws.amazon.com/sdk-for-sap-abap/v1/api/latest/index.html)于 S *AP 的AWS SDK ABAP API 参考*。 ------ 有关 S AWS DK 开发者指南和代码示例的完整列表，请参阅[将 Firehose 与 SDK 配合使用 AWS](sdk-general-information-section.md)。本主题还包括有关入门的信息以及有关先前的 SDK 版本的详细信息。