Sono disponibili altri esempi AWS SDK nel repository [AWS Doc SDK](https://github.com/awsdocs/aws-doc-sdk-examples) Examples. GitHub
Le traduzioni sono generate tramite traduzione automatica. In caso di conflitto tra il contenuto di una traduzione e la versione originale in Inglese, quest'ultima prevarrà.
# Esempi di codice per l'utilizzo di Firehose AWS SDKs
I seguenti esempi di codice mostrano come usare Amazon Data Firehose con un kit di sviluppo AWS software (SDK).
Le *azioni* sono estratti di codice da programmi più grandi e devono essere eseguite nel contesto. Sebbene le azioni mostrino come richiamare le singole funzioni del servizio, è possibile visualizzarle contestualizzate negli scenari correlati.
*Scenari*: esempi di codice che mostrano come eseguire un’attività specifica chiamando più funzioni all’interno dello stesso servizio o combinate con altri Servizi AWS.
**Altre risorse**
+ **[Guida per l’utente di Firehose](https://docs.aws.amazon.com/firehose/latest/dev/what-is-this-service.html)**: ulteriori informazioni su Firehose.
+ **[Documentazione di riferimento dell’API Firehose](https://docs.aws.amazon.com/firehose/latest/APIReference/Welcome.html)**: dettagli su tutte le azioni Firehose disponibili.
+ **[AWS Developer Center](https://aws.amazon.com/developer/code-examples/?awsf.sdk-code-examples-product=product%23kinesis-data-firehose)**: esempi di codice che puoi filtrare per categoria o per ricerca completa.
+ **[AWS Esempi SDK](https://github.com/awsdocs/aws-doc-sdk-examples)**: GitHub repository con codice completo nelle lingue preferite. Include le istruzioni su come configurare ed eseguire il codice.
**Contents**
+ [Nozioni di base](firehose_code_examples_basics.md)
+ [Azioni](firehose_code_examples_actions.md)
+ [`PutRecord`](firehose_example_firehose_PutRecord_section.md)
+ [`PutRecordBatch`](firehose_example_firehose_PutRecordBatch_section.md)
+ [Scenari](firehose_code_examples_scenarios.md)
+ [Inserire i record in Firehose](firehose_example_firehose_Scenario_PutRecords_section.md)
# Esempi di base per l'utilizzo di Firehose AWS SDKs
I seguenti esempi di codice mostrano come utilizzare le nozioni di base di Amazon Data AWS SDKs Firehose con.
**Contents**
+ [Azioni](firehose_code_examples_actions.md)
+ [`PutRecord`](firehose_example_firehose_PutRecord_section.md)
+ [`PutRecordBatch`](firehose_example_firehose_PutRecordBatch_section.md)
# Azioni per l'utilizzo di Firehose AWS SDKs
I seguenti esempi di codice mostrano come eseguire singole azioni di Firehose con. AWS SDKs Ogni esempio include un collegamento a GitHub, dove è possibile trovare le istruzioni per la configurazione e l'esecuzione del codice.
Questi estratti chiamano l’API Firehose e sono estratti di codice da programmi più grandi che devono essere eseguiti in modo contestuale. È possibile visualizzare le azioni nel contesto in [Scenari per l'utilizzo di Firehose AWS SDKs](firehose_code_examples_scenarios.md).
Gli esempi seguenti includono solo le azioni più comunemente utilizzate. Per un elenco completo, consulta la [documentazione di riferimento dell’API Amazon Data Firehose](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)
# Utilizzo `PutRecord` con un AWS SDK o una CLI
Gli esempi di codice seguenti mostrano come utilizzare `PutRecord`.
Gli esempi di operazioni sono estratti di codice da programmi più grandi e devono essere eseguiti nel contesto. È possibile visualizzare questa operazione nel contesto nel seguente esempio di codice:
+ [Inserire i record in Firehose](firehose_example_firehose_Scenario_PutRecords_section.md)
------
#### [ CLI ]
**AWS CLI**
**Come scrivere un record in un flusso**
L’esempio `put-record` seguente scrive dati in un flusso. I dati sono codificati nel formato Base64.
```
aws firehose put-record \
--delivery-stream-name my-stream \
--record '{"Data":"SGVsbG8gd29ybGQ="}'
```
Output:
```
{
"RecordId": "RjB5K/nnoGFHqwTsZlNd/TTqvjE8V5dsyXZTQn2JXrdpMTOwssyEb6nfC8fwf1whhwnItt4mvrn+gsqeK5jB7QjuLg283+Ps4Sz/j1Xujv31iDhnPdaLw4BOyM9Amv7PcCuB2079RuM0NhoakbyUymlwY8yt20G8X2420wu1jlFafhci4erAt7QhDEvpwuK8N1uOQ1EuaKZWxQHDzcG6tk1E49IPeD9k",
"Encrypted": false
}
```
Per ulteriori informazioni, consulta [Invio di dati a un flusso di distribuzione Amazon Kinesis Data Firehose](https://docs.aws.amazon.com/firehose/latest/dev/basic-write.html) nella *Guida per sviluppatori di Amazon Kinesis Data Firehose*.
+ Per i dettagli sull'API, consulta [PutRecord AWS CLI](https://awscli.amazonaws.com/v2/documentation/api/latest/reference/firehose/put-record.html)*Command Reference.*
------
#### [ Java ]
**SDK per Java 2.x**
C'è altro su GitHub. Trova l'esempio completo e scopri di più sulla configurazione e l'esecuzione nel [Repository di esempi di codice 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);
}
}
```
+ Per i dettagli sull'API, consulta la [PutRecord](https://docs.aws.amazon.com/goto/SdkForJavaV2/firehose-2015-08-04/PutRecord)sezione *AWS SDK for Java 2.x API Reference*.
------
#### [ Python ]
**SDK per Python (Boto3)**
C'è altro su GitHub. Trova l'esempio completo e scopri di più sulla configurazione e l'esecuzione nel [Repository di esempi di codice 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
```
+ Per i dettagli sull'API, consulta [PutRecord AWS](https://docs.aws.amazon.com/goto/boto3/firehose-2015-08-04/PutRecord)*SDK for Python (Boto3) API Reference*.
------
#### [ SAP ABAP ]
**SDK per SAP ABAP**
C'è altro da fare. GitHub Trova l'esempio completo e scopri di più sulla configurazione e l'esecuzione nel [Repository di esempi di codice 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.
```
+ Per i dettagli sulle API, [PutRecord](https://docs.aws.amazon.com/sdk-for-sap-abap/v1/api/latest/index.html)consulta *AWS SDK for SAP ABAP* API reference.
------
# Utilizzo `PutRecordBatch` con un AWS SDK o una CLI
Gli esempi di codice seguenti mostrano come utilizzare `PutRecordBatch`.
Gli esempi di operazioni sono estratti di codice da programmi più grandi e devono essere eseguiti nel contesto. È possibile visualizzare questa operazione nel contesto nel seguente esempio di codice:
+ [Inserire i record in Firehose](firehose_example_firehose_Scenario_PutRecords_section.md)
------
#### [ CLI ]
**AWS CLI**
**Come scrivere più record in un flusso**
L’esempio `put-record-batch` seguente scrive tre record in un flusso. I dati sono codificati nel formato Base64.
```
aws firehose put-record-batch \
--delivery-stream-name my-stream \
--records file://records.json
```
Contenuto di `myfile.json`:
```
[
{"Data": "Rmlyc3QgdGhpbmc="},
{"Data": "U2Vjb25kIHRoaW5n"},
{"Data": "VGhpcmQgdGhpbmc="}
]
```
Output:
```
{
"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"
}
]
}
```
Per ulteriori informazioni, consulta [Invio di dati a un flusso di distribuzione Amazon Kinesis Data Firehose](https://docs.aws.amazon.com/firehose/latest/dev/basic-write.html) nella *Guida per sviluppatori di Amazon Kinesis Data Firehose*.
+ Per i dettagli sull'API, consulta [PutRecordBatch AWS CLI](https://awscli.amazonaws.com/v2/documentation/api/latest/reference/firehose/put-record-batch.html)*Command Reference.*
------
#### [ Java ]
**SDK per Java 2.x**
C'è altro su GitHub. Trova l'esempio completo e scopri di più sulla configurazione e l'esecuzione nel [Repository di esempi di codice 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);
}
}
```
+ Per i dettagli sull'API, consulta la [PutRecordBatch](https://docs.aws.amazon.com/goto/SdkForJavaV2/firehose-2015-08-04/PutRecordBatch)sezione *AWS SDK for Java 2.x API Reference*.
------
#### [ Python ]
**SDK per Python (Boto3)**
C'è altro su GitHub. Trova l'esempio completo e scopri di più sulla configurazione e l'esecuzione nel [Repository di esempi di codice 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}")
```
+ Per i dettagli sull'API, consulta [PutRecordBatch AWS](https://docs.aws.amazon.com/goto/boto3/firehose-2015-08-04/PutRecordBatch)*SDK for Python (Boto3) API Reference*.
------
#### [ Rust ]
**SDK per Rust**
C'è altro da fare. GitHub Trova l'esempio completo e scopri di più sulla configurazione e l'esecuzione nel [Repository di esempi di codice 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
}
```
+ Per i dettagli sulle API, consulta la [PutRecordBatch](https://docs.rs/aws-sdk-firehose/latest/aws_sdk_firehose/client/struct.Client.html#method.put_record_batch)guida di *riferimento all'API AWS SDK for Rust*.
------
#### [ SAP ABAP ]
**SDK per SAP ABAP**
C'è altro da fare. GitHub Trova l'esempio completo e scopri di più sulla configurazione e l'esecuzione nel [Repository di esempi di codice 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.
```
+ Per i dettagli sulle API, [PutRecordBatch](https://docs.aws.amazon.com/sdk-for-sap-abap/v1/api/latest/index.html)consulta *AWS SDK for SAP ABAP* API reference.
------
# Scenari per l'utilizzo di Firehose AWS SDKs
I seguenti esempi di codice mostrano come implementare scenari comuni in Firehose con. AWS SDKs Questi scenari illustrano come eseguire attività specifiche chiamando più funzioni all’interno di Firehose o in combinazione con altri Servizi AWS. Ogni scenario include un collegamento al codice sorgente completo, dove è possibile trovare le istruzioni su come configurare ed eseguire il codice.
Gli scenari sono relativi a un livello intermedio di esperienza per aiutarti a comprendere le azioni di servizio nel contesto.
**Topics**
+ [Inserire i record in Firehose](firehose_example_firehose_Scenario_PutRecords_section.md)
# Utilizzare Amazon Data Firehose per elaborare record singoli e batch
Gli esempi di codice seguenti mostrano come utilizzare Firehose per elaborare record singoli e batch.
------
#### [ Java ]
**SDK per Java 2.x**
C'è altro da fare. GitHub Trova l'esempio completo e scopri di più sulla configurazione e l'esecuzione nel [Repository di esempi di codice AWS](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/javav2/example_code/firehose#code-examples).
Questo esempio inserisce record singoli e batch in 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());
}
}
}
```
+ Per informazioni dettagliate sull’API, consulta i seguenti argomenti nella *documentazione di riferimento dell’API AWS SDK for Java 2.x *.
+ [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 ]
**SDK per Python (Boto3)**
C'è dell'altro GitHub. Trova l'esempio completo e scopri di più sulla configurazione e l'esecuzione nel [Repository di esempi di codice AWS](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/firehose/scenarios/firehose-put-actions#code-examples).
Questo script inserisce record singoli e batch in 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()
```
Questo file contiene la configurazione per lo script precedente.
```
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()
```
+ Per informazioni dettagliate sull’API, consulta i seguenti argomenti nella *documentazione di riferimento dell’API AWS SDK per Python (Boto3)*.
+ [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)
------