文件 AWS 開發套件範例 GitHub 儲存庫中有更多可用的 [AWS SDK 範例](https://github.com/awsdocs/aws-doc-sdk-examples)。
本文為英文版的機器翻譯版本，如內容有任何歧義或不一致之處，概以英文版為準。
# Firehose 使用 AWS SDKs的動作
下列程式碼範例示範如何使用 AWS SDKs 執行個別 Firehose 動作。每個範例均包含 GitHub 的連結，您可以在連結中找到設定和執行程式碼的相關說明。
這些摘錄會呼叫 Firehose API，是必須在內容中執行之大型程式的程式碼摘錄。您可以在 [Firehose 使用 AWS SDKs案例](firehose_code_examples_scenarios.md) 中查看內容中的動作。
下列範例僅包含最常使用的動作。如需完整清單，請參閱《[Amazon Data Firehose API 參考](https://docs.aws.amazon.com/firehose/latest/APIReference/Welcome.html)》。
**Topics**
+ [`PutRecord`](firehose_example_firehose_PutRecord_section.md)
+ [`PutRecordBatch`](firehose_example_firehose_PutRecordBatch_section.md)
# `PutRecord` 搭配 AWS SDK 或 CLI 使用
下列程式碼範例示範如何使用 `PutRecord`。
動作範例是大型程式的程式碼摘錄，必須在內容中執行。您可以在下列程式碼範例的內容中看到此動作：
+ [將記錄放入 Firehose](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 ]
**SDK for Java 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 詳細資訊，請參閱《AWS SDK for Python (Boto3) API 參考》**中的 [PutRecord](https://docs.aws.amazon.com/goto/boto3/firehose-2015-08-04/PutRecord)。
------
#### [ SAP ABAP ]
**適用於 SAP ABAP 的開發套件**
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 詳細資訊，請參閱《適用於 SAP ABAP 的AWS SDK API 參考》**中的 [PutRecord](https://docs.aws.amazon.com/sdk-for-sap-abap/v1/api/latest/index.html)。
------
# `PutRecordBatch` 搭配 AWS SDK 或 CLI 使用
下列程式碼範例示範如何使用 `PutRecordBatch`。
動作範例是大型程式的程式碼摘錄，必須在內容中執行。您可以在下列程式碼範例的內容中看到此動作：
+ [將記錄放入 Firehose](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 ]
**SDK for Java 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 詳細資訊，請參閱《AWS SDK for Python (Boto3) API 參考》**中的 [PutRecordBatch](https://docs.aws.amazon.com/goto/boto3/firehose-2015-08-04/PutRecordBatch)。
------
#### [ 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 詳細資訊，請參閱《AWS SDK for Rust API 參考》**中的 [PutRecordBatch](https://docs.rs/aws-sdk-firehose/latest/aws_sdk_firehose/client/struct.Client.html#method.put_record_batch)。
------
#### [ SAP ABAP ]
**適用於 SAP ABAP 的開發套件**
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 詳細資訊，請參閱《適用於 *AWS SAP ABAP 的 SDK API 參考*》中的 [PutRecordBatch](https://docs.aws.amazon.com/sdk-for-sap-abap/v1/api/latest/index.html)。
------