文档 AWS SDK 示例 GitHub 存储库中还有更多 [S AWS DK 示例](https://github.com/awsdocs/aws-doc-sdk-examples)。
本文属于机器翻译版本。若本译文内容与英语原文存在差异，则一律以英文原文为准。
# 使用 Firehose 的代码示例 AWS SDKs
以下代码示例向您展示了如何将 Amazon Data Firehose 与 AWS 软件开发套件 (SDK) 配合使用。
*操作*是大型程序的代码摘录，必须在上下文中运行。您可以通过操作了解如何调用单个服务函数，还可以通过函数相关场景的上下文查看操作。
*场景*是向您展示如何通过在一个服务中调用多个函数或与其他 AWS 服务服务结合来完成特定任务的代码示例。
**更多资源**
+ **[ Firehose 用户指南](https://docs.aws.amazon.com/firehose/latest/dev/what-is-this-service.html)**——有关 Firehose 的更多信息。
+ **[Firehose API 参考](https://docs.aws.amazon.com/firehose/latest/APIReference/Welcome.html)**——有关所有可用的 Firehose 操作的详细信息。
+ **[AWS 开发者中心](https://aws.amazon.com/developer/code-examples/?awsf.sdk-code-examples-product=product%23kinesis-data-firehose)** — 您可以按类别或全文搜索筛选的代码示例。
+ **[AWS SDK 示例](https://github.com/awsdocs/aws-doc-sdk-examples)** — 包含首选语言完整代码的 GitHub 存储库。包括有关设置和运行代码的说明。
**Contents**
+ [基本功能](firehose_code_examples_basics.md)
+ [操作](firehose_code_examples_actions.md)
+ [`PutRecord`](firehose_example_firehose_PutRecord_section.md)
+ [`PutRecordBatch`](firehose_example_firehose_PutRecordBatch_section.md)
+ [场景](firehose_code_examples_scenarios.md)
+ [将记录放入 Firehose](firehose_example_firehose_Scenario_PutRecords_section.md)
# 使用 Firehose 的基本示例 AWS SDKs
以下代码示例展示了如何使用 Amazon Data Firehose 的基础知识。 AWS SDKs
**Contents**
+ [操作](firehose_code_examples_actions.md)
+ [`PutRecord`](firehose_example_firehose_PutRecord_section.md)
+ [`PutRecordBatch`](firehose_example_firehose_PutRecordBatch_section.md)
# 使用 Firehose 的操作 AWS SDKs
以下代码示例演示了如何使用执行各个 Firehose 操作。 AWS SDKs每个示例都包含一个指向的链接 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 ]
**适用于 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 参考*。
------
# `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 ]
**适用于 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 参考*。
------
# 使用 Firehose 的场景 AWS SDKs
以下代码示例向您展示了如何使用在 Firehose 中实现常见场景。 AWS SDKs这些场景向您展示了如何通过调用 Firehose 中的多个函数或与其他 AWS 服务结合来完成特定任务。每个场景都包含完整源代码的链接，您可以在其中找到有关如何设置和运行代码的说明。
场景以中等水平的经验为目标，可帮助您结合具体环境了解服务操作。
**Topics**
+ [将记录放入 Firehose](firehose_example_firehose_Scenario_PutRecords_section.md)
# 使用 Amazon Data Firehose 处理单个记录和批量记录
以下代码示例演示如何使用 Firehose 处理单个记录和批量记录。
------
#### [ Java ]
**适用于 Java 的 SDK 2.x**
还有更多相关信息 GitHub。在 [AWS 代码示例存储库](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/javav2/example_code/firehose#code-examples)中查找完整示例，了解如何进行设置和运行。
此示例将单个记录和批量记录放入 Firehose。
```
/**
* Amazon Firehose Scenario example using Java V2 SDK.
*
* Demonstrates individual and batch record processing,
* and monitoring Firehose delivery stream metrics.
*/
public class FirehoseScenario {
private static FirehoseClient firehoseClient;
private static CloudWatchClient cloudWatchClient;
public static void main(String[] args) {
final String usage = """
Usage:
Where:
deliveryStreamName - The Firehose delivery stream name.
""";
if (args.length != 1) {
System.out.println(usage);
return;
}
String deliveryStreamName = args[0];
try {
// Read and parse sample data.
String jsonContent = readJsonFile("sample_records.json");
ObjectMapper objectMapper = new ObjectMapper();
List> sampleData = objectMapper.readValue(jsonContent, new TypeReference<>() {});
// Process individual records.
System.out.println("Processing individual records...");
sampleData.subList(0, 100).forEach(record -> {
try {
putRecord(record, deliveryStreamName);
} catch (Exception e) {
System.err.println("Error processing record: " + e.getMessage());
}
});
// Monitor metrics.
monitorMetrics(deliveryStreamName);
// Process batch records.
System.out.println("Processing batch records...");
putRecordBatch(sampleData.subList(100, sampleData.size()), 500, deliveryStreamName);
monitorMetrics(deliveryStreamName);
} catch (Exception e) {
System.err.println("Scenario failed: " + e.getMessage());
} finally {
closeClients();
}
}
private static FirehoseClient getFirehoseClient() {
if (firehoseClient == null) {
firehoseClient = FirehoseClient.builder()
.region(Region.US_EAST_1)
.build();
}
return firehoseClient;
}
private static CloudWatchClient getCloudWatchClient() {
if (cloudWatchClient == null) {
cloudWatchClient = CloudWatchClient.builder()
.region(Region.US_EAST_1)
.build();
}
return cloudWatchClient;
}
/**
* 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);
}
}
/**
* 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);
}
}
public static void monitorMetrics(String deliveryStreamName) {
Instant endTime = Instant.now();
Instant startTime = endTime.minusSeconds(600);
List metrics = List.of("IncomingBytes", "IncomingRecords", "FailedPutCount");
metrics.forEach(metric -> monitorMetric(metric, startTime, endTime, deliveryStreamName));
}
private static void monitorMetric(String metricName, Instant startTime, Instant endTime, String deliveryStreamName) {
try {
GetMetricStatisticsRequest request = GetMetricStatisticsRequest.builder()
.namespace("AWS/Firehose")
.metricName(metricName)
.dimensions(Dimension.builder().name("DeliveryStreamName").value(deliveryStreamName).build())
.startTime(startTime)
.endTime(endTime)
.period(60)
.statistics(Statistic.SUM)
.build();
GetMetricStatisticsResponse response = getCloudWatchClient().getMetricStatistics(request);
double totalSum = response.datapoints().stream().mapToDouble(Datapoint::sum).sum();
System.out.println(metricName + ": " + totalSum);
} catch (Exception e) {
System.err.println("Failed to monitor metric " + metricName + ": " + e.getMessage());
}
}
public static String readJsonFile(String fileName) throws IOException {
try (InputStream inputStream = FirehoseScenario.class.getResourceAsStream("/" + fileName);
Scanner scanner = new Scanner(inputStream, StandardCharsets.UTF_8)) {
return scanner.useDelimiter("\\\\A").next();
} catch (Exception e) {
throw new RuntimeException("Error reading file: " + fileName, e);
}
}
private static void closeClients() {
try {
if (firehoseClient != null) firehoseClient.close();
if (cloudWatchClient != null) cloudWatchClient.close();
} catch (Exception e) {
System.err.println("Error closing clients: " + e.getMessage());
}
}
}
```
+ 有关 API 详细信息，请参阅《AWS SDK for Java 2.x API Reference》**中的以下主题。
+ [PutRecord](https://docs.aws.amazon.com/goto/SdkForJavaV2/firehose-2015-08-04/PutRecord)
+ [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/scenarios/firehose-put-actions#code-examples)中查找完整示例，了解如何进行设置和运行。
此脚本将单个记录和批量记录放入 Firehose。
```
import json
import logging
import random
from datetime import datetime, timedelta
import backoff
import boto3
from config import get_config
def load_sample_data(path: str) -> dict:
"""
Load sample data from a JSON file.
Args:
path (str): The file path to the JSON file containing sample data.
Returns:
dict: The loaded sample data as a dictionary.
"""
with open(path, "r") as f:
return json.load(f)
# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
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
@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}")
def get_metric_statistics(
self,
metric_name: str,
start_time: datetime,
end_time: datetime,
period: int,
statistics: list = ["Sum"],
) -> list:
"""
Retrieve metric statistics from CloudWatch.
Args:
metric_name (str): The name of the metric.
start_time (datetime): The start time for the metric statistics.
end_time (datetime): The end time for the metric statistics.
period (int): The granularity, in seconds, of the returned data points.
statistics (list): A list of statistics to retrieve. Default is ['Sum'].
Returns:
list: List of datapoints containing the metric statistics.
"""
response = self.cloudwatch.get_metric_statistics(
Namespace="AWS/Firehose",
MetricName=metric_name,
Dimensions=[
{"Name": "DeliveryStreamName", "Value": self.delivery_stream_name},
],
StartTime=start_time,
EndTime=end_time,
Period=period,
Statistics=statistics,
)
return response["Datapoints"]
def monitor_metrics(self):
"""
Monitor Firehose metrics for the last 5 minutes.
This method retrieves and logs the 'IncomingBytes', 'IncomingRecords', and 'FailedPutCount' metrics
from CloudWatch for the last 5 minutes.
"""
end_time = datetime.utcnow()
start_time = end_time - timedelta(minutes=10)
period = int((end_time - start_time).total_seconds())
metrics = {
"IncomingBytes": self.get_metric_statistics(
"IncomingBytes", start_time, end_time, period
),
"IncomingRecords": self.get_metric_statistics(
"IncomingRecords", start_time, end_time, period
),
"FailedPutCount": self.get_metric_statistics(
"FailedPutCount", start_time, end_time, period
),
}
for metric, datapoints in metrics.items():
if datapoints:
total_sum = sum(datapoint["Sum"] for datapoint in datapoints)
if metric == "IncomingBytes":
logger.info(
f"{metric}: {round(total_sum)} ({total_sum / (1024 * 1024):.2f} MB)"
)
else:
logger.info(f"{metric}: {round(total_sum)}")
else:
logger.info(f"No data found for {metric} over the last 5 minutes")
def _create_record_entry(self, record: dict) -> dict:
"""
Create a record entry for Firehose.
Args:
record (dict): The data record to be sent.
Returns:
dict: The record entry formatted for Firehose.
Raises:
Exception: If a simulated network error occurs.
"""
if random.random() < 0.2:
raise Exception("Simulated network error")
elif random.random() < 0.1:
return {"Data": '{"malformed": "data"'}
else:
return {"Data": json.dumps(record)}
def _log_response(self, response: dict, entry: dict):
"""
Log the response from Firehose.
Args:
response (dict): The response from the Firehose put_record API call.
entry (dict): The record entry that was sent.
"""
if response["ResponseMetadata"]["HTTPStatusCode"] == 200:
logger.info(f"Sent record: {entry}")
else:
logger.info(f"Fail record: {entry}")
def _log_batch_response(self, response: dict, batch_size: int):
"""
Log the batch response from Firehose.
Args:
response (dict): The response from the Firehose put_record_batch API call.
batch_size (int): The number of records in the batch.
"""
if response.get("FailedPutCount", 0) > 0:
logger.info(
f'Failed to send {response["FailedPutCount"]} records in batch of {batch_size}'
)
else:
logger.info(f"Successfully sent batch of {batch_size} records")
if __name__ == "__main__":
config = get_config()
data = load_sample_data(config.sample_data_file)
client = FirehoseClient(config)
# Process the first 100 sample network records
for record in data[:100]:
try:
client.put_record(record)
except Exception as e:
logger.info(f"Put record failed after retries and backoff: {e}")
client.monitor_metrics()
# Process remaining records using the batch method
try:
client.put_record_batch(data[100:])
except Exception as e:
logger.info(f"Put record batch failed after retries and backoff: {e}")
client.monitor_metrics()
```
此文件包含上述脚本的配置。
```
class Config:
def __init__(self):
self.delivery_stream_name = "ENTER YOUR DELIVERY STREAM NAME HERE"
self.region = "us-east-1"
self.sample_data_file = (
"../../../../../scenarios/features/firehose/resources/sample_records.json"
)
def get_config():
return Config()
```
+ 有关 API 详细信息，请参阅《AWS SDK for Python (Boto3) API Reference》**中的以下主题。
+ [PutRecord](https://docs.aws.amazon.com/goto/boto3/firehose-2015-08-04/PutRecord)
+ [PutRecordBatch](https://docs.aws.amazon.com/goto/boto3/firehose-2015-08-04/PutRecordBatch)
------