Hay más ejemplos de AWS SDK disponibles en el GitHub repositorio de [ejemplos de AWS Doc SDK](https://github.com/awsdocs/aws-doc-sdk-examples).
Las traducciones son generadas a través de traducción automática. En caso de conflicto entre la traducción y la version original de inglés, prevalecerá la version en inglés.
# Ejemplos de código para el uso de Firehose AWS SDKs
Los siguientes ejemplos de código muestran cómo utilizar Amazon Data Firehose con un kit de desarrollo de AWS software (SDK).
Las *acciones* son extractos de código de programas más grandes y deben ejecutarse en contexto. Mientras las acciones muestran cómo llamar a las distintas funciones de servicio, es posible ver las acciones en contexto en los escenarios relacionados.
Los *escenarios* son ejemplos de código que muestran cómo llevar a cabo una tarea específica a través de llamadas a varias funciones dentro del servicio o combinado con otros Servicios de AWS.
**Más recursos**
+ **[Guía del usuario de Firehose](https://docs.aws.amazon.com/firehose/latest/dev/what-is-this-service.html)**: más información sobre Firehose.
+ **[Referencia de la API de Firehose](https://docs.aws.amazon.com/firehose/latest/APIReference/Welcome.html)**: información sobre todas las acciones disponibles en Firehose.
+ **[AWS Centro de desarrolladores](https://aws.amazon.com/developer/code-examples/?awsf.sdk-code-examples-product=product%23kinesis-data-firehose)**: ejemplos de código que puede filtrar por categoría o por búsqueda de texto completo.
+ **[AWS Ejemplos de SDK](https://github.com/awsdocs/aws-doc-sdk-examples)**: GitHub repositorio con código completo en los idiomas preferidos. Incluye instrucciones para configurar y ejecutar el código.
**Contents**
+ [Conceptos básicos](firehose_code_examples_basics.md)
+ [Acciones](firehose_code_examples_actions.md)
+ [`PutRecord`](firehose_example_firehose_PutRecord_section.md)
+ [`PutRecordBatch`](firehose_example_firehose_PutRecordBatch_section.md)
+ [Escenarios](firehose_code_examples_scenarios.md)
+ [Colocar los registros en Firehose](firehose_example_firehose_Scenario_PutRecords_section.md)
# Ejemplos básicos de uso de Firehose AWS SDKs
Los siguientes ejemplos de código muestran cómo utilizar los conceptos básicos de Amazon Data Firehose con. AWS SDKs
**Contents**
+ [Acciones](firehose_code_examples_actions.md)
+ [`PutRecord`](firehose_example_firehose_PutRecord_section.md)
+ [`PutRecordBatch`](firehose_example_firehose_PutRecordBatch_section.md)
# Acciones para el uso de Firehose AWS SDKs
Los siguientes ejemplos de código muestran cómo realizar acciones individuales de Firehose con. AWS SDKs Cada ejemplo incluye un enlace a GitHub, donde puede encontrar instrucciones para configurar y ejecutar el código.
Estos fragmentos llaman a la API de Firehose y son fragmentos de código de programas más grandes que deben ejecutarse en contexto. Puede ver las acciones en contexto en [Escenarios para el uso de Firehose AWS SDKs](firehose_code_examples_scenarios.md).
Los siguientes ejemplos incluyen solo las acciones que se utilizan con mayor frecuencia. Para ver una lista completa, consulte la [Referencia de la API de 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)
# Úselo `PutRecord` con un AWS SDK o CLI
Los siguientes ejemplos de código muestran cómo utilizar `PutRecord`.
Los ejemplos de acciones son extractos de código de programas más grandes y deben ejecutarse en contexto. Puede ver esta acción en contexto en el siguiente ejemplo de código:
+ [Colocar los registros en Firehose](firehose_example_firehose_Scenario_PutRecords_section.md)
------
#### [ CLI ]
**AWS CLI**
**Escritura de un registro en un flujo**
En el siguiente ejemplo de `put-record`, se escriben datos en un flujo. Los datos se codifican en formato Base64.
```
aws firehose put-record \
--delivery-stream-name my-stream \
--record '{"Data":"SGVsbG8gd29ybGQ="}'
```
Salida:
```
{
"RecordId": "RjB5K/nnoGFHqwTsZlNd/TTqvjE8V5dsyXZTQn2JXrdpMTOwssyEb6nfC8fwf1whhwnItt4mvrn+gsqeK5jB7QjuLg283+Ps4Sz/j1Xujv31iDhnPdaLw4BOyM9Amv7PcCuB2079RuM0NhoakbyUymlwY8yt20G8X2420wu1jlFafhci4erAt7QhDEvpwuK8N1uOQ1EuaKZWxQHDzcG6tk1E49IPeD9k",
"Encrypted": false
}
```
Para obtener más información, consulte [Envío de una secuencia de entrega de Amazon Kinesis Data Firehose](https://docs.aws.amazon.com/firehose/latest/dev/basic-write.html) en la *Guía para desarrolladores de Amazon Kinesis Data Firehose*.
+ Para obtener más información sobre la API, consulte [PutRecord](https://awscli.amazonaws.com/v2/documentation/api/latest/reference/firehose/put-record.html)la *Referencia de AWS CLI comandos*.
------
#### [ Java ]
**SDK para Java 2.x**
Hay más información al respecto GitHub. Busque el ejemplo completo y aprenda a configurar y ejecutar en el [Repositorio de ejemplos de código de 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);
}
}
```
+ Para obtener más información sobre la API, consulta [PutRecord](https://docs.aws.amazon.com/goto/SdkForJavaV2/firehose-2015-08-04/PutRecord)la *Referencia AWS SDK for Java 2.x de la API*.
------
#### [ Python ]
**SDK para Python (Boto3)**
Hay más información al respecto GitHub. Busque el ejemplo completo y aprenda a configurar y ejecutar en el [Repositorio de ejemplos de código de 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
```
+ Para obtener más información sobre la API, consulta [PutRecord](https://docs.aws.amazon.com/goto/boto3/firehose-2015-08-04/PutRecord)la *AWS Referencia de API de SDK for Python (Boto3*).
------
#### [ SAP ABAP ]
**SDK para SAP ABAP**
Hay más información al respecto. GitHub Busque el ejemplo completo y aprenda a configurar y ejecutar en el [Repositorio de ejemplos de código de 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.
```
+ Para obtener más información sobre la API, consulte [PutRecord](https://docs.aws.amazon.com/sdk-for-sap-abap/v1/api/latest/index.html)la *referencia sobre la API ABAP del AWS SDK para SAP*.
------
# Úselo `PutRecordBatch` con un AWS SDK o CLI
Los siguientes ejemplos de código muestran cómo utilizar `PutRecordBatch`.
Los ejemplos de acciones son extractos de código de programas más grandes y deben ejecutarse en contexto. Puede ver esta acción en contexto en el siguiente ejemplo de código:
+ [Colocar los registros en Firehose](firehose_example_firehose_Scenario_PutRecords_section.md)
------
#### [ CLI ]
**AWS CLI**
**Para escribir varios registros en un flujo**
En el siguiente ejemplo `put-record-batch`, se escriben tres registros en una secuencia. Los datos se codifican en formato Base64.
```
aws firehose put-record-batch \
--delivery-stream-name my-stream \
--records file://records.json
```
Contenido de `myfile.json`:
```
[
{"Data": "Rmlyc3QgdGhpbmc="},
{"Data": "U2Vjb25kIHRoaW5n"},
{"Data": "VGhpcmQgdGhpbmc="}
]
```
Salida:
```
{
"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"
}
]
}
```
Para obtener más información, consulte [Envío de una secuencia de entrega de Amazon Kinesis Data Firehose](https://docs.aws.amazon.com/firehose/latest/dev/basic-write.html) en la *Guía para desarrolladores de Amazon Kinesis Data Firehose*.
+ Para obtener más información sobre la API, consulte [PutRecordBatch](https://awscli.amazonaws.com/v2/documentation/api/latest/reference/firehose/put-record-batch.html)la *Referencia de AWS CLI comandos*.
------
#### [ Java ]
**SDK para Java 2.x**
Hay más información al respecto GitHub. Busque el ejemplo completo y aprenda a configurar y ejecutar en el [Repositorio de ejemplos de código de 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);
}
}
```
+ Para obtener más información sobre la API, consulta [PutRecordBatch](https://docs.aws.amazon.com/goto/SdkForJavaV2/firehose-2015-08-04/PutRecordBatch)la *Referencia AWS SDK for Java 2.x de la API*.
------
#### [ Python ]
**SDK para Python (Boto3)**
Hay más información al respecto GitHub. Busque el ejemplo completo y aprenda a configurar y ejecutar en el [Repositorio de ejemplos de código de 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}")
```
+ Para obtener más información sobre la API, consulta [PutRecordBatch](https://docs.aws.amazon.com/goto/boto3/firehose-2015-08-04/PutRecordBatch)la *AWS Referencia de API de SDK for Python (Boto3*).
------
#### [ Rust ]
**SDK para Rust**
Hay más información al respecto. GitHub Busque el ejemplo completo y aprenda a configurar y ejecutar en el [Repositorio de ejemplos de código de 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
}
```
+ Para obtener más información sobre la API, consulta [PutRecordBatch](https://docs.rs/aws-sdk-firehose/latest/aws_sdk_firehose/client/struct.Client.html#method.put_record_batch)la *referencia sobre la API de AWS SDK para Rust*.
------
#### [ SAP ABAP ]
**SDK para SAP ABAP**
Hay más información al respecto GitHub. Busque el ejemplo completo y aprenda a configurar y ejecutar en el [Repositorio de ejemplos de código de 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.
```
+ Para obtener más información sobre la API, consulte [PutRecordBatch](https://docs.aws.amazon.com/sdk-for-sap-abap/v1/api/latest/index.html)la *referencia sobre la API ABAP del AWS SDK para SAP*.
------
# Escenarios para el uso de Firehose AWS SDKs
Los siguientes ejemplos de código muestran cómo implementar escenarios comunes en Firehose con. AWS SDKs Estas situaciones muestran cómo llevar a cabo tareas específicas con llamadas a varias funciones dentro de Firehose o en combinación con otros Servicios de AWS. En cada escenario se incluye un enlace al código fuente completo, con instrucciones de configuración y ejecución del código.
Los escenarios requieren un nivel intermedio de experiencia para entender las acciones de servicio en su contexto.
**Topics**
+ [Colocar los registros en Firehose](firehose_example_firehose_Scenario_PutRecords_section.md)
# Utilice Amazon Data Firehose para procesar registros individuales y por lotes
Los siguientes ejemplos de código muestran cómo usar Firehose para procesar registros individuales y por lotes.
------
#### [ Java ]
**SDK para Java 2.x**
Hay más en marcha GitHub. Busque el ejemplo completo y aprenda a configurar y ejecutar en el [Repositorio de ejemplos de código de AWS](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/javav2/example_code/firehose#code-examples).
Este ejemplo coloca los registros individuales y por lotes en 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());
}
}
}
```
+ Para obtener detalles sobre la API, consulte los siguientes temas en la *Referencia de la API de 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 para Python (Boto3)**
Hay más información GitHub. Busque el ejemplo completo y aprenda a configurar y ejecutar en el [Repositorio de ejemplos de código de AWS](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/firehose/scenarios/firehose-put-actions#code-examples).
Este script coloca los registros individuales y por lotes en 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()
```
Este archivo contiene la configuración del script anterior.
```
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()
```
+ Para obtener información sobre la API, consulte los siguientes temas en la *Referencia de la API de AWS SDK para 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)
------