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 Kinesis AWS SDKs
I seguenti esempi di codice mostrano come usare Amazon Kinesis con un kit di sviluppo AWS software (SDK).
*Nozioni di base*: esempi di codice che mostrano come eseguire le operazioni essenziali all’interno di un servizio.
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.
**Altre risorse**
+ **[Guida per gli sviluppatori di Kinesis](https://docs.aws.amazon.com/streams/latest/dev/introduction.html)**: ulteriori informazioni su Kinesis.
+ **[Documentazione di riferimento dell’API Kinesis](https://docs.aws.amazon.com/kinesis/latest/APIReference/Welcome.html)**: dettagli su tutte le azioni Kinesis disponibili.
+ **[AWS Developer Center](https://aws.amazon.com/developer/code-examples/?awsf.sdk-code-examples-product=product%23kinesis)**: 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](kinesis_code_examples_basics.md)
+ [Informazioni di base](kinesis_example_kinesis_Scenario_GettingStarted_section.md)
+ [Azioni](kinesis_code_examples_actions.md)
+ [`AddTagsToStream`](kinesis_example_kinesis_AddTagsToStream_section.md)
+ [`CreateStream`](kinesis_example_kinesis_CreateStream_section.md)
+ [`DeleteStream`](kinesis_example_kinesis_DeleteStream_section.md)
+ [`DeregisterStreamConsumer`](kinesis_example_kinesis_DeregisterStreamConsumer_section.md)
+ [`DescribeStream`](kinesis_example_kinesis_DescribeStream_section.md)
+ [`GetRecords`](kinesis_example_kinesis_GetRecords_section.md)
+ [`GetShardIterator`](kinesis_example_kinesis_GetShardIterator_section.md)
+ [`ListStreamConsumers`](kinesis_example_kinesis_ListStreamConsumers_section.md)
+ [`ListStreams`](kinesis_example_kinesis_ListStreams_section.md)
+ [`ListTagsForStream`](kinesis_example_kinesis_ListTagsForStream_section.md)
+ [`PutRecord`](kinesis_example_kinesis_PutRecord_section.md)
+ [`PutRecords`](kinesis_example_kinesis_PutRecords_section.md)
+ [`RegisterStreamConsumer`](kinesis_example_kinesis_RegisterStreamConsumer_section.md)
+ [Esempi serverless](kinesis_code_examples_serverless_examples.md)
+ [Invocare una funzione Lambda da un trigger Kinesis](kinesis_example_serverless_Kinesis_Lambda_section.md)
+ [Segnalazione di errori di elementi batch per funzioni Lambda con un trigger Kinesis](kinesis_example_serverless_Kinesis_Lambda_batch_item_failures_section.md)
# Esempi di base per l'utilizzo di Kinesis AWS SDKs
I seguenti esempi di codice mostrano come utilizzare le nozioni di base di Amazon AWS SDKs Kinesis con.
**Contents**
+ [Informazioni di base](kinesis_example_kinesis_Scenario_GettingStarted_section.md)
+ [Azioni](kinesis_code_examples_actions.md)
+ [`AddTagsToStream`](kinesis_example_kinesis_AddTagsToStream_section.md)
+ [`CreateStream`](kinesis_example_kinesis_CreateStream_section.md)
+ [`DeleteStream`](kinesis_example_kinesis_DeleteStream_section.md)
+ [`DeregisterStreamConsumer`](kinesis_example_kinesis_DeregisterStreamConsumer_section.md)
+ [`DescribeStream`](kinesis_example_kinesis_DescribeStream_section.md)
+ [`GetRecords`](kinesis_example_kinesis_GetRecords_section.md)
+ [`GetShardIterator`](kinesis_example_kinesis_GetShardIterator_section.md)
+ [`ListStreamConsumers`](kinesis_example_kinesis_ListStreamConsumers_section.md)
+ [`ListStreams`](kinesis_example_kinesis_ListStreams_section.md)
+ [`ListTagsForStream`](kinesis_example_kinesis_ListTagsForStream_section.md)
+ [`PutRecord`](kinesis_example_kinesis_PutRecord_section.md)
+ [`PutRecords`](kinesis_example_kinesis_PutRecords_section.md)
+ [`RegisterStreamConsumer`](kinesis_example_kinesis_RegisterStreamConsumer_section.md)
# Scopri le basi di Kinesis con un SDK AWS
L’esempio di codice seguente mostra come:
+ Creare un flusso e inserirvi un record.
+ Creare un iteratore di shard.
+ Leggere il record, quindi eseguire la pulizia delle risorse.
------
#### [ 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/kns#code-examples).
```
DATA lo_stream_describe_result TYPE REF TO /aws1/cl_knsdescrstreamoutput.
DATA lo_stream_description TYPE REF TO /aws1/cl_knsstreamdescription.
DATA lo_sharditerator TYPE REF TO /aws1/cl_knsgetsharditerator01.
DATA lo_record_result TYPE REF TO /aws1/cl_knsputrecordoutput.
"Create stream."
TRY.
lo_kns->createstream(
iv_streamname = iv_stream_name
iv_shardcount = iv_shard_count ).
MESSAGE 'Stream created.' TYPE 'I'.
CATCH /aws1/cx_knsinvalidargumentex.
MESSAGE 'The specified argument was not valid.' TYPE 'E'.
CATCH /aws1/cx_knslimitexceededex.
MESSAGE 'The request processing has failed because of a limit exceeded exception.' TYPE 'E'.
CATCH /aws1/cx_knsresourceinuseex.
MESSAGE 'The request processing has failed because the resource is in use.' TYPE 'E'.
ENDTRY.
"Wait for stream to becomes active."
lo_stream_describe_result = lo_kns->describestream( iv_streamname = iv_stream_name ).
lo_stream_description = lo_stream_describe_result->get_streamdescription( ).
WHILE lo_stream_description->get_streamstatus( ) <> 'ACTIVE'.
IF sy-index = 30.
EXIT. "maximum 5 minutes"
ENDIF.
WAIT UP TO 10 SECONDS.
lo_stream_describe_result = lo_kns->describestream( iv_streamname = iv_stream_name ).
lo_stream_description = lo_stream_describe_result->get_streamdescription( ).
ENDWHILE.
"Create record."
TRY.
lo_record_result = lo_kns->putrecord(
iv_streamname = iv_stream_name
iv_data = iv_data
iv_partitionkey = iv_partition_key ).
MESSAGE 'Record created.' TYPE 'I'.
CATCH /aws1/cx_knsinvalidargumentex.
MESSAGE 'The specified argument was not valid.' TYPE 'E'.
CATCH /aws1/cx_knskmsaccessdeniedex.
MESSAGE 'You do not have permission to perform this AWS KMS action.' TYPE 'E'.
CATCH /aws1/cx_knskmsdisabledex.
MESSAGE 'KMS key used is disabled.' TYPE 'E'.
CATCH /aws1/cx_knskmsinvalidstateex.
MESSAGE 'KMS key used is in an invalid state. ' TYPE 'E'.
CATCH /aws1/cx_knskmsnotfoundex.
MESSAGE 'KMS key used is not found.' TYPE 'E'.
CATCH /aws1/cx_knskmsoptinrequired.
MESSAGE 'KMS key option is required.' TYPE 'E'.
CATCH /aws1/cx_knskmsthrottlingex.
MESSAGE 'The rate of requests to AWS KMS is exceeding the request quotas.' TYPE 'E'.
CATCH /aws1/cx_knsprovthruputexcdex.
MESSAGE 'The request rate for the stream is too high, or the requested data is too large for the available throughput.' TYPE 'E'.
CATCH /aws1/cx_knsresourcenotfoundex.
MESSAGE 'Resource being accessed is not found.' TYPE 'E'.
ENDTRY.
"Create a shard iterator in order to read the record."
TRY.
lo_sharditerator = lo_kns->getsharditerator(
iv_shardid = lo_record_result->get_shardid( )
iv_sharditeratortype = iv_sharditeratortype
iv_streamname = iv_stream_name ).
MESSAGE 'Shard iterator created.' TYPE 'I'.
CATCH /aws1/cx_knsinvalidargumentex.
MESSAGE 'The specified argument was not valid.' TYPE 'E'.
CATCH /aws1/cx_knsprovthruputexcdex.
MESSAGE 'The request rate for the stream is too high, or the requested data is too large for the available throughput.' TYPE 'E'.
CATCH /aws1/cx_sgmresourcenotfound.
MESSAGE 'Resource being accessed is not found.' TYPE 'E'.
ENDTRY.
"Read the record."
TRY.
oo_result = lo_kns->getrecords( " oo_result is returned for testing purposes. "
iv_sharditerator = lo_sharditerator->get_sharditerator( ) ).
MESSAGE 'Shard iterator created.' TYPE 'I'.
CATCH /aws1/cx_knsexpirediteratorex.
MESSAGE 'Iterator expired.' TYPE 'E'.
CATCH /aws1/cx_knsinvalidargumentex.
MESSAGE 'The specified argument was not valid.' TYPE 'E'.
CATCH /aws1/cx_knskmsaccessdeniedex.
MESSAGE 'You do not have permission to perform this AWS KMS action.' TYPE 'E'.
CATCH /aws1/cx_knskmsdisabledex.
MESSAGE 'KMS key used is disabled.' TYPE 'E'.
CATCH /aws1/cx_knskmsinvalidstateex.
MESSAGE 'KMS key used is in an invalid state. ' TYPE 'E'.
CATCH /aws1/cx_knskmsnotfoundex.
MESSAGE 'KMS key used is not found.' TYPE 'E'.
CATCH /aws1/cx_knskmsoptinrequired.
MESSAGE 'KMS key option is required.' TYPE 'E'.
CATCH /aws1/cx_knskmsthrottlingex.
MESSAGE 'The rate of requests to AWS KMS is exceeding the request quotas.' TYPE 'E'.
CATCH /aws1/cx_knsprovthruputexcdex.
MESSAGE 'The request rate for the stream is too high, or the requested data is too large for the available throughput.' TYPE 'E'.
CATCH /aws1/cx_knsresourcenotfoundex.
MESSAGE 'Resource being accessed is not found.' TYPE 'E'.
ENDTRY.
"Delete stream."
TRY.
lo_kns->deletestream(
iv_streamname = iv_stream_name ).
MESSAGE 'Stream deleted.' TYPE 'I'.
CATCH /aws1/cx_knslimitexceededex.
MESSAGE 'The request processing has failed because of a limit exceeded exception.' TYPE 'E'.
CATCH /aws1/cx_knsresourceinuseex.
MESSAGE 'The request processing has failed because the resource is in use.' TYPE 'E'.
ENDTRY.
```
+ Per informazioni dettagliate sull’API, consulta gli argomenti seguenti nella *documentazione di riferimento dell’API AWS SDK per SAP ABAP*.
+ [CreateStream](https://docs.aws.amazon.com/sdk-for-sap-abap/v1/api/latest/index.html)
+ [DeleteStream](https://docs.aws.amazon.com/sdk-for-sap-abap/v1/api/latest/index.html)
+ [GetRecords](https://docs.aws.amazon.com/sdk-for-sap-abap/v1/api/latest/index.html)
+ [GetShardIterator](https://docs.aws.amazon.com/sdk-for-sap-abap/v1/api/latest/index.html)
+ [PutRecord](https://docs.aws.amazon.com/sdk-for-sap-abap/v1/api/latest/index.html)
------
# Azioni per l'utilizzo di Kinesis AWS SDKs
I seguenti esempi di codice mostrano come eseguire singole azioni Kinesis con. AWS SDKs Ogni esempio include un collegamento a GitHub, dove puoi trovare le istruzioni per la configurazione e l'esecuzione del codice.
Gli esempi seguenti includono solo le azioni più comunemente utilizzate. Per un elenco completo, consulta la [documentazione di riferimento dell’API Amazon Kinesis](https://docs.aws.amazon.com/kinesis/latest/APIReference/Welcome.html).
**Topics**
+ [`AddTagsToStream`](kinesis_example_kinesis_AddTagsToStream_section.md)
+ [`CreateStream`](kinesis_example_kinesis_CreateStream_section.md)
+ [`DeleteStream`](kinesis_example_kinesis_DeleteStream_section.md)
+ [`DeregisterStreamConsumer`](kinesis_example_kinesis_DeregisterStreamConsumer_section.md)
+ [`DescribeStream`](kinesis_example_kinesis_DescribeStream_section.md)
+ [`GetRecords`](kinesis_example_kinesis_GetRecords_section.md)
+ [`GetShardIterator`](kinesis_example_kinesis_GetShardIterator_section.md)
+ [`ListStreamConsumers`](kinesis_example_kinesis_ListStreamConsumers_section.md)
+ [`ListStreams`](kinesis_example_kinesis_ListStreams_section.md)
+ [`ListTagsForStream`](kinesis_example_kinesis_ListTagsForStream_section.md)
+ [`PutRecord`](kinesis_example_kinesis_PutRecord_section.md)
+ [`PutRecords`](kinesis_example_kinesis_PutRecords_section.md)
+ [`RegisterStreamConsumer`](kinesis_example_kinesis_RegisterStreamConsumer_section.md)
# Utilizzo `AddTagsToStream` con un AWS SDK o una CLI
Gli esempi di codice seguenti mostrano come utilizzare `AddTagsToStream`.
------
#### [ .NET ]
**SDK per .NET**
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/dotnetv3/Kinesis#code-examples).
```
using System;
using System.Collections.Generic;
using System.Threading.Tasks;
using Amazon.Kinesis;
using Amazon.Kinesis.Model;
///
/// This example shows how to apply key/value pairs to an Amazon Kinesis
/// stream.
///
public class TagStream
{
public static async Task Main()
{
IAmazonKinesis client = new AmazonKinesisClient();
string streamName = "AmazonKinesisStream";
var tags = new Dictionary
{
{ "Project", "Sample Kinesis Project" },
{ "Application", "Sample Kinesis App" },
};
var success = await ApplyTagsToStreamAsync(client, streamName, tags);
if (success)
{
Console.WriteLine($"Taggs successfully added to {streamName}.");
}
else
{
Console.WriteLine("Tags were not added to the stream.");
}
}
///
/// Applies the set of tags to the named Kinesis stream.
///
/// The initialized Kinesis client.
/// The name of the Kinesis stream to which
/// the tags will be attached.
/// A sictionary containing key/value pairs which
/// will be used to create the Kinesis tags.
/// A Boolean value which represents the success or failure
/// of AddTagsToStreamAsync.
public static async Task ApplyTagsToStreamAsync(
IAmazonKinesis client,
string streamName,
Dictionary tags)
{
var request = new AddTagsToStreamRequest
{
StreamName = streamName,
Tags = tags,
};
var response = await client.AddTagsToStreamAsync(request);
return response.HttpStatusCode == System.Net.HttpStatusCode.OK;
}
}
```
+ Per i dettagli sull'API, [AddTagsToStream](https://docs.aws.amazon.com/goto/DotNetSDKV3/kinesis-2013-12-02/AddTagsToStream)consulta *AWS SDK per .NET API Reference*.
------
#### [ CLI ]
**AWS CLI**
**Come aggiungere tag a un flusso di dati**
L’esempio `add-tags-to-stream` seguente assegna un tag con la chiave `samplekey` e il valore `example` al flusso specificato.
```
aws kinesis add-tags-to-stream \
--stream-name samplestream \
--tags samplekey=example
```
Questo comando non produce alcun output.
Per ulteriori informazioni, consulta [Tagging dei flussi](https://docs.aws.amazon.com/streams/latest/dev/tagging.html) nella *Guida per gli sviluppatori di Flusso di dati Amazon Kinesis*.
+ Per i dettagli sull'API, consulta [AddTagsToStream AWS CLI](https://awscli.amazonaws.com/v2/documentation/api/latest/reference/kinesis/add-tags-to-stream.html)*Command Reference*.
------
# Utilizzo `CreateStream` con un AWS SDK o una CLI
Gli esempi di codice seguenti mostrano come utilizzare `CreateStream`.
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:
+ [Informazioni di base](kinesis_example_kinesis_Scenario_GettingStarted_section.md)
------
#### [ .NET ]
**SDK per .NET**
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/dotnetv3/Kinesis#code-examples).
```
using System;
using System.Threading.Tasks;
using Amazon.Kinesis;
using Amazon.Kinesis.Model;
///
/// This example shows how to create a new Amazon Kinesis stream.
///
public class CreateStream
{
public static async Task Main()
{
IAmazonKinesis client = new AmazonKinesisClient();
string streamName = "AmazonKinesisStream";
int shardCount = 1;
var success = await CreateNewStreamAsync(client, streamName, shardCount);
if (success)
{
Console.WriteLine($"The stream, {streamName} successfully created.");
}
}
///
/// Creates a new Kinesis stream.
///
/// An initialized Kinesis client.
/// The name for the new stream.
/// The number of shards the new stream will
/// use. The throughput of the stream is a function of the number of
/// shards; more shards are required for greater provisioned
/// throughput.
/// A Boolean value indicating whether the stream was created.
public static async Task CreateNewStreamAsync(IAmazonKinesis client, string streamName, int shardCount)
{
var request = new CreateStreamRequest
{
StreamName = streamName,
ShardCount = shardCount,
};
var response = await client.CreateStreamAsync(request);
return response.HttpStatusCode == System.Net.HttpStatusCode.OK;
}
}
```
+ Per i dettagli sull'API, [CreateStream](https://docs.aws.amazon.com/goto/DotNetSDKV3/kinesis-2013-12-02/CreateStream)consulta *AWS SDK per .NET API Reference*.
------
#### [ CLI ]
**AWS CLI**
**Come creare un flusso di dati**
L’esempio `create-stream` seguente crea un flusso di dati denominato samplestream con 3 shard.
```
aws kinesis create-stream \
--stream-name samplestream \
--shard-count 3
```
Questo comando non produce alcun output.
Per ulteriori informazioni, consulta [Creazione di un flusso](https://docs.aws.amazon.com/streams/latest/dev/kinesis-using-sdk-java-create-stream.html) nella *Guida per gli sviluppatori di Flusso di dati Amazon Kinesis*.
+ Per i dettagli sull'API, consulta [CreateStream AWS CLI](https://awscli.amazonaws.com/v2/documentation/api/latest/reference/kinesis/create-stream.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/kinesis#code-examples).
```
import software.amazon.awssdk.regions.Region;
import software.amazon.awssdk.services.kinesis.KinesisClient;
import software.amazon.awssdk.services.kinesis.model.CreateStreamRequest;
import software.amazon.awssdk.services.kinesis.model.KinesisException;
/**
* Before running this Java V2 code example, set up your development
* environment, including your credentials.
*
* For more information, see the following documentation topic:
*
* https://docs.aws.amazon.com/sdk-for-java/latest/developer-guide/get-started.html
*/
public class CreateDataStream {
public static void main(String[] args) {
final String usage = """
Usage:
Where:
streamName - The Amazon Kinesis data stream (for example, StockTradeStream).
""";
if (args.length != 1) {
System.out.println(usage);
System.exit(1);
}
String streamName = args[0];
Region region = Region.US_EAST_1;
KinesisClient kinesisClient = KinesisClient.builder()
.region(region)
.build();
createStream(kinesisClient, streamName);
System.out.println("Done");
kinesisClient.close();
}
public static void createStream(KinesisClient kinesisClient, String streamName) {
try {
CreateStreamRequest streamReq = CreateStreamRequest.builder()
.streamName(streamName)
.shardCount(1)
.build();
kinesisClient.createStream(streamReq);
} catch (KinesisException e) {
System.err.println(e.getMessage());
System.exit(1);
}
}
}
```
+ Per i dettagli sull'API, [CreateStream](https://docs.aws.amazon.com/goto/SdkForJavaV2/kinesis-2013-12-02/CreateStream)consulta *AWS SDK for Java 2.x API Reference*.
------
#### [ PowerShell ]
**Strumenti per PowerShell V4**
**Esempio 1: crea un nuovo flusso. Per impostazione predefinita, questo cmdlet non restituisce alcun output, pertanto viene aggiunto lo PassThru switch - per restituire il valore fornito al StreamName parametro - per un uso successivo.**
```
$streamName = New-KINStream -StreamName "mystream" -ShardCount 1 -PassThru
```
+ Per i dettagli sull'API, vedere [CreateStream](https://docs.aws.amazon.com/powershell/v4/reference)in *AWS Strumenti per PowerShell Cmdlet Reference* (V4).
**Strumenti per V5 PowerShell **
**Esempio 1: crea un nuovo flusso.**
```
New-KINStream -StreamName "mystream" -ShardCount 1
```
+ Per i dettagli sull'API, vedere [CreateStream](https://docs.aws.amazon.com/powershell/v5/reference)in *AWS Strumenti per PowerShell Cmdlet Reference (*V5).
------
#### [ 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/kinesis#code-examples).
```
class KinesisStream:
"""Encapsulates a Kinesis stream."""
def __init__(self, kinesis_client):
"""
:param kinesis_client: A Boto3 Kinesis client.
"""
self.kinesis_client = kinesis_client
self.name = None
self.details = None
self.stream_exists_waiter = kinesis_client.get_waiter("stream_exists")
def create(self, name, wait_until_exists=True):
"""
Creates a stream.
:param name: The name of the stream.
:param wait_until_exists: When True, waits until the service reports that
the stream exists, then queries for its metadata.
"""
try:
self.kinesis_client.create_stream(StreamName=name, ShardCount=1)
self.name = name
logger.info("Created stream %s.", name)
if wait_until_exists:
logger.info("Waiting until exists.")
self.stream_exists_waiter.wait(StreamName=name)
self.describe(name)
except ClientError:
logger.exception("Couldn't create stream %s.", name)
raise
```
+ Per i dettagli sull'API, consulta [CreateStream AWS](https://docs.aws.amazon.com/goto/boto3/kinesis-2013-12-02/CreateStream)*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/kinesis#code-examples).
```
async fn make_stream(client: &Client, stream: &str) -> Result<(), Error> {
client
.create_stream()
.stream_name(stream)
.shard_count(4)
.send()
.await?;
println!("Created stream");
Ok(())
}
```
+ Per i dettagli sulle API, consulta il *riferimento [CreateStream](https://docs.rs/aws-sdk-kinesis/latest/aws_sdk_kinesis/client/struct.Client.html#method.create_stream)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/kns#code-examples).
```
TRY.
lo_kns->createstream(
iv_streamname = iv_stream_name
iv_shardcount = iv_shard_count ).
MESSAGE 'Stream created.' TYPE 'I'.
CATCH /aws1/cx_knsinvalidargumentex.
MESSAGE 'The specified argument was not valid.' TYPE 'E'.
CATCH /aws1/cx_knslimitexceededex.
MESSAGE 'The request processing has failed because of a limit exceed exception.' TYPE 'E'.
CATCH /aws1/cx_knsresourceinuseex.
MESSAGE 'The request processing has failed because the resource is in use.' TYPE 'E'.
ENDTRY.
```
+ Per i dettagli sulle API, [CreateStream](https://docs.aws.amazon.com/sdk-for-sap-abap/v1/api/latest/index.html)consulta *AWS SDK for SAP ABAP* API reference.
------
# Utilizzo `DeleteStream` con un AWS SDK o una CLI
Gli esempi di codice seguenti mostrano come utilizzare `DeleteStream`.
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:
+ [Informazioni di base](kinesis_example_kinesis_Scenario_GettingStarted_section.md)
------
#### [ .NET ]
**SDK per .NET**
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/dotnetv3/Kinesis#code-examples).
```
using System;
using System.Threading.Tasks;
using Amazon.Kinesis;
using Amazon.Kinesis.Model;
///
/// Shows how to delete an Amazon Kinesis stream.
///
public class DeleteStream
{
public static async Task Main()
{
IAmazonKinesis client = new AmazonKinesisClient();
string streamName = "AmazonKinesisStream";
var success = await DeleteStreamAsync(client, streamName);
if (success)
{
Console.WriteLine($"Stream, {streamName} successfully deleted.");
}
else
{
Console.WriteLine("Stream not deleted.");
}
}
///
/// Deletes a Kinesis stream.
///
/// An initialized Kinesis client object.
/// The name of the string to delete.
/// A Boolean value representing the success of the operation.
public static async Task DeleteStreamAsync(IAmazonKinesis client, string streamName)
{
// If EnforceConsumerDeletion is true, any consumers
// of this stream will also be deleted. If it is set
// to false and this stream has any consumers, the
// call will fail with a ResourceInUseException.
var request = new DeleteStreamRequest
{
StreamName = streamName,
EnforceConsumerDeletion = true,
};
var response = await client.DeleteStreamAsync(request);
return response.HttpStatusCode == System.Net.HttpStatusCode.OK;
}
}
```
+ Per i dettagli sull'API, [DeleteStream](https://docs.aws.amazon.com/goto/DotNetSDKV3/kinesis-2013-12-02/DeleteStream)consulta *AWS SDK per .NET API Reference*.
------
#### [ CLI ]
**AWS CLI**
**Come eliminare un flusso di dati**
L’esempio `delete-stream` seguente elimina il flusso di dati specificato.
```
aws kinesis delete-stream \
--stream-name samplestream
```
Questo comando non produce alcun output.
Per ulteriori informazioni, consulta [Eliminazione di un flusso](https://docs.aws.amazon.com/streams/latest/dev/kinesis-using-sdk-java-delete-stream.html) nella *Guida per gli sviluppatori di Flusso di dati Amazon Kinesis*.
+ Per i dettagli sull'API, consulta [DeleteStream AWS CLI](https://awscli.amazonaws.com/v2/documentation/api/latest/reference/kinesis/delete-stream.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/kinesis#code-examples).
```
import software.amazon.awssdk.regions.Region;
import software.amazon.awssdk.services.kinesis.KinesisClient;
import software.amazon.awssdk.services.kinesis.model.DeleteStreamRequest;
import software.amazon.awssdk.services.kinesis.model.KinesisException;
/**
* Before running this Java V2 code example, set up your development
* environment, including your credentials.
*
* For more information, see the following documentation topic:
*
* https://docs.aws.amazon.com/sdk-for-java/latest/developer-guide/get-started.html
*/
public class DeleteDataStream {
public static void main(String[] args) {
final String usage = """
Usage:
Where:
streamName - The Amazon Kinesis data stream (for example, StockTradeStream)
""";
if (args.length != 1) {
System.out.println(usage);
System.exit(1);
}
String streamName = args[0];
Region region = Region.US_EAST_1;
KinesisClient kinesisClient = KinesisClient.builder()
.region(region)
.build();
deleteStream(kinesisClient, streamName);
kinesisClient.close();
System.out.println("Done");
}
public static void deleteStream(KinesisClient kinesisClient, String streamName) {
try {
DeleteStreamRequest delStream = DeleteStreamRequest.builder()
.streamName(streamName)
.build();
kinesisClient.deleteStream(delStream);
} catch (KinesisException e) {
System.err.println(e.getMessage());
System.exit(1);
}
}
}
```
+ Per i dettagli sull'API, [DeleteStream](https://docs.aws.amazon.com/goto/SdkForJavaV2/kinesis-2013-12-02/DeleteStream)consulta *AWS SDK for Java 2.x API Reference*.
------
#### [ PowerShell ]
**Strumenti per PowerShell V4**
**Esempio 1: elimina il flusso specificato. Viene richiesta una conferma prima dell’esecuzione del comando. Per disabilitare la richiesta di conferma, utilizza il parametro -Force.**
```
Remove-KINStream -StreamName "mystream"
```
+ Per i dettagli sull'API, vedere [DeleteStream](https://docs.aws.amazon.com/powershell/v4/reference)in *AWS Strumenti per PowerShell Cmdlet Reference (*V4).
**Strumenti per V5 PowerShell **
**Esempio 1: elimina il flusso specificato. Viene richiesta una conferma prima dell’esecuzione del comando. Per disabilitare la richiesta di conferma, utilizza il parametro -Force.**
```
Remove-KINStream -StreamName "mystream"
```
+ Per i dettagli sull'API, vedere [DeleteStream](https://docs.aws.amazon.com/powershell/v5/reference)in *AWS Strumenti per PowerShell Cmdlet Reference (*V5).
------
#### [ 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/kinesis#code-examples).
```
class KinesisStream:
"""Encapsulates a Kinesis stream."""
def __init__(self, kinesis_client):
"""
:param kinesis_client: A Boto3 Kinesis client.
"""
self.kinesis_client = kinesis_client
self.name = None
self.details = None
self.stream_exists_waiter = kinesis_client.get_waiter("stream_exists")
def delete(self):
"""
Deletes a stream.
"""
try:
self.kinesis_client.delete_stream(StreamName=self.name)
self._clear()
logger.info("Deleted stream %s.", self.name)
except ClientError:
logger.exception("Couldn't delete stream %s.", self.name)
raise
```
+ Per i dettagli sull'API, consulta [DeleteStream AWS](https://docs.aws.amazon.com/goto/boto3/kinesis-2013-12-02/DeleteStream)*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/kinesis#code-examples).
```
async fn remove_stream(client: &Client, stream: &str) -> Result<(), Error> {
client.delete_stream().stream_name(stream).send().await?;
println!("Deleted stream.");
Ok(())
}
```
+ Per i dettagli sulle API, consulta il *riferimento [DeleteStream](https://docs.rs/aws-sdk-kinesis/latest/aws_sdk_kinesis/client/struct.Client.html#method.delete_stream)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/kns#code-examples).
```
TRY.
lo_kns->deletestream(
iv_streamname = iv_stream_name ).
MESSAGE 'Stream deleted.' TYPE 'I'.
CATCH /aws1/cx_knslimitexceededex.
MESSAGE 'The request processing has failed because of a limit exceed exception.' TYPE 'E'.
CATCH /aws1/cx_knsresourceinuseex.
MESSAGE 'The request processing has failed because the resource is in use.' TYPE 'E'.
ENDTRY.
```
+ Per i dettagli sulle API, [DeleteStream](https://docs.aws.amazon.com/sdk-for-sap-abap/v1/api/latest/index.html)consulta *AWS SDK for SAP ABAP* API reference.
------
# Utilizzo `DeregisterStreamConsumer` con un AWS SDK o una CLI
Gli esempi di codice seguenti mostrano come utilizzare `DeregisterStreamConsumer`.
------
#### [ .NET ]
**SDK per .NET**
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/dotnetv3/Kinesis#code-examples).
```
using System;
using System.Threading.Tasks;
using Amazon.Kinesis;
using Amazon.Kinesis.Model;
///
/// Shows how to deregister a consumer from an Amazon Kinesis stream.
///
public class DeregisterConsumer
{
public static async Task Main(string[] args)
{
IAmazonKinesis client = new AmazonKinesisClient();
string streamARN = "arn:aws:kinesis:us-west-2:000000000000:stream/AmazonKinesisStream";
string consumerName = "CONSUMER_NAME";
string consumerARN = "arn:aws:kinesis:us-west-2:000000000000:stream/AmazonKinesisStream/consumer/CONSUMER_NAME:000000000000";
var success = await DeregisterConsumerAsync(client, streamARN, consumerARN, consumerName);
if (success)
{
Console.WriteLine($"{consumerName} successfully deregistered.");
}
else
{
Console.WriteLine($"{consumerName} was not successfully deregistered.");
}
}
///
/// Deregisters a consumer from a Kinesis stream.
///
/// An initialized Kinesis client object.
/// The ARN of a Kinesis stream.
/// The ARN of the consumer.
/// The name of the consumer.
/// A Boolean value representing the success of the operation.
public static async Task DeregisterConsumerAsync(
IAmazonKinesis client,
string streamARN,
string consumerARN,
string consumerName)
{
var request = new DeregisterStreamConsumerRequest
{
StreamARN = streamARN,
ConsumerARN = consumerARN,
ConsumerName = consumerName,
};
var response = await client.DeregisterStreamConsumerAsync(request);
return response.HttpStatusCode == System.Net.HttpStatusCode.OK;
}
}
```
+ Per i dettagli sull'API, [DeregisterStreamConsumer](https://docs.aws.amazon.com/goto/DotNetSDKV3/kinesis-2013-12-02/DeregisterStreamConsumer)consulta *AWS SDK per .NET API Reference*.
------
#### [ CLI ]
**AWS CLI**
**Come annullare la registrazione di un consumer del flusso di dati**
L’esempio `deregister-stream-consumer` seguente annulla la registrazione del consumer specificato dal flusso di dati specificato.
```
aws kinesis deregister-stream-consumer \
--stream-arn arn:aws:kinesis:us-west-2:123456789012:stream/samplestream \
--consumer-name KinesisConsumerApplication
```
Questo comando non produce alcun output.
Per ulteriori informazioni, consulta [Sviluppo di applicazioni consumer con fan-out avanzato con l’API di Flusso dati Kinesis](https://docs.aws.amazon.com/streams/latest/dev/building-enhanced-consumers-api.html) nella *Guida per gli sviluppatori di Flusso di dati Amazon Kinesis*.
+ Per i dettagli sull'API, consulta [DeregisterStreamConsumer AWS CLI](https://awscli.amazonaws.com/v2/documentation/api/latest/reference/kinesis/deregister-stream-consumer.html)*Command Reference*.
------
# Utilizzo `DescribeStream` con un AWS SDK o una CLI
Gli esempi di codice seguenti mostrano come utilizzare `DescribeStream`.
------
#### [ CLI ]
**AWS CLI**
**Come descrivere un flusso di dati**
L’esempio `describe-stream` seguente restituisce i dettagli del flusso di dati specificato.
```
aws kinesis describe-stream \
--stream-name samplestream
```
Output:
```
{
"StreamDescription": {
"Shards": [
{
"ShardId": "shardId-000000000000",
"HashKeyRange": {
"StartingHashKey": "0",
"EndingHashKey": "113427455640312821154458202477256070484"
},
"SequenceNumberRange": {
"StartingSequenceNumber": "49600871682957036442365024926191073437251060580128653314"
}
},
{
"ShardId": "shardId-000000000001",
"HashKeyRange": {
"StartingHashKey": "113427455640312821154458202477256070485",
"EndingHashKey": "226854911280625642308916404954512140969"
},
"SequenceNumberRange": {
"StartingSequenceNumber": "49600871682979337187563555549332609155523708941634633746"
}
},
{
"ShardId": "shardId-000000000002",
"HashKeyRange": {
"StartingHashKey": "226854911280625642308916404954512140970",
"EndingHashKey": "340282366920938463463374607431768211455"
},
"SequenceNumberRange": {
"StartingSequenceNumber": "49600871683001637932762086172474144873796357303140614178"
}
}
],
"StreamARN": "arn:aws:kinesis:us-west-2:123456789012:stream/samplestream",
"StreamName": "samplestream",
"StreamStatus": "ACTIVE",
"RetentionPeriodHours": 24,
"EnhancedMonitoring": [
{
"ShardLevelMetrics": []
}
],
"EncryptionType": "NONE",
"KeyId": null,
"StreamCreationTimestamp": 1572297168.0
}
}
```
Per ulteriori informazioni, consulta [Creazione e gestione dei flussi](https://docs.aws.amazon.com/streams/latest/dev/working-with-streams.html) nella *Guida per gli sviluppatori di Flusso di dati Amazon Kinesis*.
+ Per i dettagli sull'API, consulta [DescribeStream AWS CLI](https://awscli.amazonaws.com/v2/documentation/api/latest/reference/kinesis/describe-stream.html)*Command Reference.*
------
#### [ PowerShell ]
**Strumenti per PowerShell V4**
**Esempio 1: restituisce i dettagli del flusso specificato.**
```
Get-KINStream -StreamName "mystream"
```
**Output:**
```
HasMoreShards : False
RetentionPeriodHours : 24
Shards : {}
StreamARN : arn:aws:kinesis:us-west-2:123456789012:stream/mystream
StreamName : mystream
StreamStatus : ACTIVE
```
+ Per i dettagli sull'API, vedere [DescribeStream](https://docs.aws.amazon.com/powershell/v4/reference)in *AWS Strumenti per PowerShell Cmdlet Reference (*V4).
**Strumenti per V5 PowerShell **
**Esempio 1: restituisce i dettagli del flusso specificato.**
```
Get-KINStream -StreamName "mystream"
```
**Output:**
```
HasMoreShards : False
RetentionPeriodHours : 24
Shards : {}
StreamARN : arn:aws:kinesis:us-west-2:123456789012:stream/mystream
StreamName : mystream
StreamStatus : ACTIVE
```
+ Per i dettagli sull'API, vedere [DescribeStream](https://docs.aws.amazon.com/powershell/v5/reference)in *AWS Strumenti per PowerShell Cmdlet Reference (*V5).
------
#### [ 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/kinesis#code-examples).
```
class KinesisStream:
"""Encapsulates a Kinesis stream."""
def __init__(self, kinesis_client):
"""
:param kinesis_client: A Boto3 Kinesis client.
"""
self.kinesis_client = kinesis_client
self.name = None
self.details = None
self.stream_exists_waiter = kinesis_client.get_waiter("stream_exists")
def describe(self, name):
"""
Gets metadata about a stream.
:param name: The name of the stream.
:return: Metadata about the stream.
"""
try:
response = self.kinesis_client.describe_stream(StreamName=name)
self.name = name
self.details = response["StreamDescription"]
logger.info("Got stream %s.", name)
except ClientError:
logger.exception("Couldn't get %s.", name)
raise
else:
return self.details
```
+ Per i dettagli sull'API, consulta [DescribeStream AWS](https://docs.aws.amazon.com/goto/boto3/kinesis-2013-12-02/DescribeStream)*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/kinesis#code-examples).
```
async fn show_stream(client: &Client, stream: &str) -> Result<(), Error> {
let resp = client.describe_stream().stream_name(stream).send().await?;
let desc = resp.stream_description.unwrap();
println!("Stream description:");
println!(" Name: {}:", desc.stream_name());
println!(" Status: {:?}", desc.stream_status());
println!(" Open shards: {:?}", desc.shards.len());
println!(" Retention (hours): {}", desc.retention_period_hours());
println!(" Encryption: {:?}", desc.encryption_type.unwrap());
Ok(())
}
```
+ Per i dettagli sulle API, consulta il *riferimento [DescribeStream](https://docs.rs/aws-sdk-kinesis/latest/aws_sdk_kinesis/client/struct.Client.html#method.describe_stream)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/kns#code-examples).
```
TRY.
oo_result = lo_kns->describestream(
iv_streamname = iv_stream_name ).
DATA(lt_stream_description) = oo_result->get_streamdescription( ).
MESSAGE 'Streams retrieved.' TYPE 'I'.
CATCH /aws1/cx_knslimitexceededex.
MESSAGE 'The request processing has failed because of a limit exceed exception.' TYPE 'E'.
CATCH /aws1/cx_knsresourcenotfoundex.
MESSAGE 'Resource being accessed is not found.' TYPE 'E'.
ENDTRY.
```
+ Per i dettagli sulle API, [DescribeStream](https://docs.aws.amazon.com/sdk-for-sap-abap/v1/api/latest/index.html)consulta *AWS SDK for SAP ABAP* API reference.
------
# Utilizzo `GetRecords` con un AWS SDK o una CLI
Gli esempi di codice seguenti mostrano come utilizzare `GetRecords`.
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:
+ [Informazioni di base](kinesis_example_kinesis_Scenario_GettingStarted_section.md)
------
#### [ CLI ]
**AWS CLI**
**Come ottenere record da uno shard**
L’esempio `get-records` seguente ottiene i record di dati dallo shard di un flusso di dati Kinesis mediante l’iterazione dei shard specificata.
```
aws kinesis get-records \
--shard-iterator AAAAAAAAAAF7/0mWD7IuHj1yGv/TKuNgx2ukD5xipCY4cy4gU96orWwZwcSXh3K9tAmGYeOZyLZrvzzeOFVf9iN99hUPw/w/b0YWYeehfNvnf1DYt5XpDJghLKr3DzgznkTmMymDP3R+3wRKeuEw6/kdxY2yKJH0veaiekaVc4N2VwK/GvaGP2Hh9Fg7N++q0Adg6fIDQPt4p8RpavDbk+A4sL9SWGE1
```
Output:
```
{
"Records": [],
"MillisBehindLatest": 80742000
}
```
Per ulteriori informazioni, consulta [Developing Consumers Using the Kinesis Data Streams API AWS with the SDK for Java](https://docs.aws.amazon.com/streams/latest/dev/developing-consumers-with-sdk.html) nella Amazon *Kinesis Data Streams* Developer Guide.
+ *Per i dettagli sull'API, consulta Command Reference. [GetRecords](https://awscli.amazonaws.com/v2/documentation/api/latest/reference/kinesis/get-records.html)AWS CLI *
------
#### [ 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/kinesis#code-examples).
```
import software.amazon.awssdk.core.SdkBytes;
import software.amazon.awssdk.regions.Region;
import software.amazon.awssdk.services.kinesis.KinesisClient;
import software.amazon.awssdk.services.kinesis.model.DescribeStreamResponse;
import software.amazon.awssdk.services.kinesis.model.DescribeStreamRequest;
import software.amazon.awssdk.services.kinesis.model.Shard;
import software.amazon.awssdk.services.kinesis.model.GetShardIteratorRequest;
import software.amazon.awssdk.services.kinesis.model.GetShardIteratorResponse;
import software.amazon.awssdk.services.kinesis.model.Record;
import software.amazon.awssdk.services.kinesis.model.GetRecordsRequest;
import software.amazon.awssdk.services.kinesis.model.GetRecordsResponse;
import java.util.ArrayList;
import java.util.List;
/**
* Before running this Java V2 code example, set up your development
* environment, including your credentials.
*
* For more information, see the following documentation topic:
*
* https://docs.aws.amazon.com/sdk-for-java/latest/developer-guide/get-started.html
*/
public class GetRecords {
public static void main(String[] args) {
final String usage = """
Usage:
Where:
streamName - The Amazon Kinesis data stream to read from (for example, StockTradeStream).
""";
if (args.length != 1) {
System.out.println(usage);
System.exit(1);
}
String streamName = args[0];
Region region = Region.US_EAST_1;
KinesisClient kinesisClient = KinesisClient.builder()
.region(region)
.build();
getStockTrades(kinesisClient, streamName);
kinesisClient.close();
}
public static void getStockTrades(KinesisClient kinesisClient, String streamName) {
String shardIterator;
String lastShardId = null;
DescribeStreamRequest describeStreamRequest = DescribeStreamRequest.builder()
.streamName(streamName)
.build();
List shards = new ArrayList<>();
DescribeStreamResponse streamRes;
do {
streamRes = kinesisClient.describeStream(describeStreamRequest);
shards.addAll(streamRes.streamDescription().shards());
if (shards.size() > 0) {
lastShardId = shards.get(shards.size() - 1).shardId();
}
} while (streamRes.streamDescription().hasMoreShards());
GetShardIteratorRequest itReq = GetShardIteratorRequest.builder()
.streamName(streamName)
.shardIteratorType("TRIM_HORIZON")
.shardId(lastShardId)
.build();
GetShardIteratorResponse shardIteratorResult = kinesisClient.getShardIterator(itReq);
shardIterator = shardIteratorResult.shardIterator();
// Continuously read data records from shard.
List records;
// Create new GetRecordsRequest with existing shardIterator.
// Set maximum records to return to 1000.
GetRecordsRequest recordsRequest = GetRecordsRequest.builder()
.shardIterator(shardIterator)
.limit(1000)
.build();
GetRecordsResponse result = kinesisClient.getRecords(recordsRequest);
// Put result into record list. Result may be empty.
records = result.records();
// Print records
for (Record record : records) {
SdkBytes byteBuffer = record.data();
System.out.printf("Seq No: %s - %s%n", record.sequenceNumber(), new String(byteBuffer.asByteArray()));
}
}
}
```
+ Per i dettagli sull'API, [GetRecords](https://docs.aws.amazon.com/goto/SdkForJavaV2/kinesis-2013-12-02/GetRecords)consulta *AWS SDK for Java 2.x API Reference*.
------
#### [ PowerShell ]
**Strumenti per PowerShell V4**
**Esempio 1: questo esempio mostra come restituire ed estrarre dati da una serie di uno o più record. L'iteratore supplierd to Get-KINRecord determina la posizione iniziale dei record da restituire, che in questo esempio vengono acquisiti in una variabile, \$1records. È quindi possibile accedere a ogni record indicizzando la raccolta \$1records. Supponendo che i dati nel record siano testo con codifica UTF-8, il comando finale mostra come estrarre i dati dall'oggetto e restituirli MemoryStream come testo alla console.**
```
$records
$records = Get-KINRecord -ShardIterator "AAAAAAAAAAGIc....9VnbiRNaP"
```
**Output:**
```
MillisBehindLatest NextShardIterator Records
------------------ ----------------- -------
0 AAAAAAAAAAERNIq...uDn11HuUs {Key1, Key2}
```
```
$records.Records[0]
```
**Output:**
```
ApproximateArrivalTimestamp Data PartitionKey SequenceNumber
--------------------------- ---- ------------ --------------
3/7/2016 5:14:33 PM System.IO.MemoryStream Key1 4955986459776...931586
```
```
[Text.Encoding]::UTF8.GetString($records.Records[0].Data.ToArray())
```
**Output:**
```
test data from string
```
+ Per i dettagli sull'API, vedere [GetRecords](https://docs.aws.amazon.com/powershell/v4/reference)in *AWS Strumenti per PowerShell Cmdlet* Reference (V4).
**Strumenti per V5 PowerShell **
**Esempio 1: questo esempio mostra come restituire ed estrarre dati da una serie di uno o più record. L'iteratore supplierd to Get-KINRecord determina la posizione iniziale dei record da restituire, che in questo esempio vengono acquisiti in una variabile, \$1records. È quindi possibile accedere a ogni record indicizzando la raccolta \$1records. Supponendo che i dati nel record siano testo con codifica UTF-8, il comando finale mostra come estrarre i dati dall'oggetto e restituirli MemoryStream come testo alla console.**
```
$records
$records = Get-KINRecord -ShardIterator "AAAAAAAAAAGIc....9VnbiRNaP"
```
**Output:**
```
MillisBehindLatest NextShardIterator Records
------------------ ----------------- -------
0 AAAAAAAAAAERNIq...uDn11HuUs {Key1, Key2}
```
```
$records.Records[0]
```
**Output:**
```
ApproximateArrivalTimestamp Data PartitionKey SequenceNumber
--------------------------- ---- ------------ --------------
3/7/2016 5:14:33 PM System.IO.MemoryStream Key1 4955986459776...931586
```
```
[Text.Encoding]::UTF8.GetString($records.Records[0].Data.ToArray())
```
**Output:**
```
test data from string
```
+ Per i dettagli sull'API, vedere [GetRecords](https://docs.aws.amazon.com/powershell/v5/reference)in *AWS Strumenti per PowerShell Cmdlet* Reference (V5).
------
#### [ 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/kinesis#code-examples).
```
class KinesisStream:
"""Encapsulates a Kinesis stream."""
def __init__(self, kinesis_client):
"""
:param kinesis_client: A Boto3 Kinesis client.
"""
self.kinesis_client = kinesis_client
self.name = None
self.details = None
self.stream_exists_waiter = kinesis_client.get_waiter("stream_exists")
def get_records(self, max_records):
"""
Gets records from the stream. This function is a generator that first gets
a shard iterator for the stream, then uses the shard iterator to get records
in batches from the stream. The shard iterator can be accessed through the
'details' property, which is populated using the 'describe' function of this class.
Each batch of records is yielded back to the caller until the specified
maximum number of records has been retrieved.
:param max_records: The maximum number of records to retrieve.
:return: Yields the current batch of retrieved records.
"""
try:
response = self.kinesis_client.get_shard_iterator(
StreamName=self.name,
ShardId=self.details["Shards"][0]["ShardId"],
ShardIteratorType="LATEST",
)
shard_iter = response["ShardIterator"]
record_count = 0
while record_count < max_records:
response = self.kinesis_client.get_records(
ShardIterator=shard_iter, Limit=10
)
shard_iter = response["NextShardIterator"]
records = response["Records"]
logger.info("Got %s records.", len(records))
record_count += len(records)
yield records
except ClientError:
logger.exception("Couldn't get records from stream %s.", self.name)
raise
def describe(self, name):
"""
Gets metadata about a stream.
:param name: The name of the stream.
:return: Metadata about the stream.
"""
try:
response = self.kinesis_client.describe_stream(StreamName=name)
self.name = name
self.details = response["StreamDescription"]
logger.info("Got stream %s.", name)
except ClientError:
logger.exception("Couldn't get %s.", name)
raise
else:
return self.details
```
+ Per i dettagli sull'API, consulta [GetRecords AWS](https://docs.aws.amazon.com/goto/boto3/kinesis-2013-12-02/GetRecords)*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/kns#code-examples).
```
TRY.
oo_result = lo_kns->getrecords( " oo_result is returned for testing purposes. "
iv_sharditerator = iv_shard_iterator ).
DATA(lt_records) = oo_result->get_records( ).
MESSAGE 'Record retrieved.' TYPE 'I'.
CATCH /aws1/cx_knsexpirediteratorex.
MESSAGE 'Iterator expired.' TYPE 'E'.
CATCH /aws1/cx_knsinvalidargumentex.
MESSAGE 'The specified argument was not valid.' TYPE 'E'.
CATCH /aws1/cx_knskmsaccessdeniedex.
MESSAGE 'You do not have permission to perform this AWS KMS action.' TYPE 'E'.
CATCH /aws1/cx_knskmsdisabledex.
MESSAGE 'KMS key used is disabled.' TYPE 'E'.
CATCH /aws1/cx_knskmsinvalidstateex.
MESSAGE 'KMS key used is in an invalid state. ' TYPE 'E'.
CATCH /aws1/cx_knskmsnotfoundex.
MESSAGE 'KMS key used is not found.' TYPE 'E'.
CATCH /aws1/cx_knskmsoptinrequired.
MESSAGE 'KMS key option is required.' TYPE 'E'.
CATCH /aws1/cx_knskmsthrottlingex.
MESSAGE 'The rate of requests to AWS KMS is exceeding the request quotas.' TYPE 'E'.
CATCH /aws1/cx_knsprovthruputexcdex.
MESSAGE 'The request rate for the stream is too high, or the requested data is too large for the available throughput.' TYPE 'E'.
CATCH /aws1/cx_knsresourcenotfoundex.
MESSAGE 'Resource being accessed is not found.' TYPE 'E'.
ENDTRY.
```
+ Per i dettagli sulle API, [GetRecords](https://docs.aws.amazon.com/sdk-for-sap-abap/v1/api/latest/index.html)consulta *AWS SDK for SAP ABAP* API reference.
------
# Utilizzare `GetShardIterator` con una CLI
Gli esempi di codice seguenti mostrano come utilizzare `GetShardIterator`.
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:
+ [Informazioni di base](kinesis_example_kinesis_Scenario_GettingStarted_section.md)
------
#### [ CLI ]
**AWS CLI**
**Come ottenere un iteratore di shard**
L’esempio `get-shard-iterator` seguente utilizza il tipo di iteratore dello shard `AT_SEQUENCE_NUMBER` e genera un iteratore dello shard per avviare la lettura dei record di dati esattamente dalla posizione indicata dal numero di sequenza specificato.
```
aws kinesis get-shard-iterator \
--stream-name samplestream \
--shard-id shardId-000000000001 \
--shard-iterator-type LATEST
```
Output:
```
{
"ShardIterator": "AAAAAAAAAAFEvJjIYI+3jw/4aqgH9FifJ+n48XWTh/IFIsbILP6o5eDueD39NXNBfpZ10WL5K6ADXk8w+5H+Qhd9cFA9k268CPXCz/kebq1TGYI7Vy+lUkA9BuN3xvATxMBGxRY3zYK05gqgvaIRn94O8SqeEqwhigwZxNWxID3Ej7YYYcxQi8Q/fIrCjGAy/n2r5Z9G864YpWDfN9upNNQAR/iiOWKs"
}
```
Per ulteriori informazioni, consulta [Developing Consumers Using the Kinesis Data Streams API AWS with the SDK for Java](https://docs.aws.amazon.com/streams/latest/dev/developing-consumers-with-sdk.html) nella Amazon *Kinesis Data Streams* Developer Guide.
+ *Per i dettagli sull'API, consulta Command Reference. [GetShardIterator](https://awscli.amazonaws.com/v2/documentation/api/latest/reference/kinesis/get-shard-iterator.html)AWS CLI *
------
#### [ PowerShell ]
**Strumenti per PowerShell V4**
**Esempio 1: restituisce un’iterazione di shard per lo shard e la posizione iniziale specificati. I dettagli degli identificatori degli shard e dei numeri di sequenza possono essere ottenuti dall'output del Get-KINStream cmdlet, facendo riferimento all'insieme Shards dell'oggetto stream restituito. L'iteratore restituito può essere utilizzato con il Get-KINRecord cmdlet per estrarre i record di dati nello shard.**
```
Get-KINShardIterator -StreamName "mystream" -ShardId "shardId-000000000000" -ShardIteratorType AT_SEQUENCE_NUMBER -StartingSequenceNumber "495598645..."
```
**Output:**
```
AAAAAAAAAAGIc....9VnbiRNaP
```
+ Per i dettagli sull'API, vedere [GetShardIterator](https://docs.aws.amazon.com/powershell/v4/reference)in *AWS Strumenti per PowerShell Cmdlet* Reference (V4).
**Strumenti per V5 PowerShell **
**Esempio 1: restituisce un’iterazione di shard per lo shard e la posizione iniziale specificati. I dettagli degli identificatori degli shard e dei numeri di sequenza possono essere ottenuti dall'output del Get-KINStream cmdlet, facendo riferimento all'insieme Shards dell'oggetto stream restituito. L'iteratore restituito può essere utilizzato con il Get-KINRecord cmdlet per estrarre i record di dati nello shard.**
```
Get-KINShardIterator -StreamName "mystream" -ShardId "shardId-000000000000" -ShardIteratorType AT_SEQUENCE_NUMBER -StartingSequenceNumber "495598645..."
```
**Output:**
```
AAAAAAAAAAGIc....9VnbiRNaP
```
+ Per i dettagli sull'API, vedere [GetShardIterator](https://docs.aws.amazon.com/powershell/v5/reference)in *AWS Strumenti per PowerShell Cmdlet* Reference (V5).
------
# Utilizzare `ListStreamConsumers` con un SDK AWS
Il seguente esempio di codice mostra come utilizzare`ListStreamConsumers`.
------
#### [ .NET ]
**SDK per .NET**
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/dotnetv3/Kinesis#code-examples).
```
using System;
using System.Collections.Generic;
using System.Threading.Tasks;
using Amazon.Kinesis;
using Amazon.Kinesis.Model;
///
/// List the consumers of an Amazon Kinesis stream.
///
public class ListConsumers
{
public static async Task Main()
{
IAmazonKinesis client = new AmazonKinesisClient();
string streamARN = "arn:aws:kinesis:us-east-2:000000000000:stream/AmazonKinesisStream";
int maxResults = 10;
var consumers = await ListConsumersAsync(client, streamARN, maxResults);
if (consumers.Count > 0)
{
consumers
.ForEach(c => Console.WriteLine($"Name: {c.ConsumerName} ARN: {c.ConsumerARN}"));
}
else
{
Console.WriteLine("No consumers found.");
}
}
///
/// Retrieve a list of the consumers for a Kinesis stream.
///
/// An initialized Kinesis client object.
/// The ARN of the stream for which we want to
/// retrieve a list of clients.
/// The maximum number of results to return.
/// A list of Consumer objects.
public static async Task> ListConsumersAsync(IAmazonKinesis client, string streamARN, int maxResults)
{
var request = new ListStreamConsumersRequest
{
StreamARN = streamARN,
MaxResults = maxResults,
};
var response = await client.ListStreamConsumersAsync(request);
return response.Consumers;
}
}
```
+ Per i dettagli sull'API, [ListStreamConsumers](https://docs.aws.amazon.com/goto/DotNetSDKV3/kinesis-2013-12-02/ListStreamConsumers)consulta *AWS SDK per .NET API Reference*.
------
# Utilizzo `ListStreams` con un AWS SDK o una CLI
Gli esempi di codice seguenti mostrano come utilizzare `ListStreams`.
------
#### [ .NET ]
**SDK per .NET**
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/dotnetv3/Kinesis#code-examples).
```
using System;
using System.Collections.Generic;
using System.Threading.Tasks;
using Amazon.Kinesis;
using Amazon.Kinesis.Model;
///
/// Retrieves and displays a list of existing Amazon Kinesis streams.
///
public class ListStreams
{
public static async Task Main(string[] args)
{
IAmazonKinesis client = new AmazonKinesisClient();
var response = await client.ListStreamsAsync(new ListStreamsRequest());
List streamNames = response.StreamNames;
if (streamNames.Count > 0)
{
streamNames
.ForEach(s => Console.WriteLine($"Stream name: {s}"));
}
else
{
Console.WriteLine("No streams were found.");
}
}
}
```
+ Per i dettagli sull'API, [ListStreams](https://docs.aws.amazon.com/goto/DotNetSDKV3/kinesis-2013-12-02/ListStreams)consulta *AWS SDK per .NET API Reference*.
------
#### [ CLI ]
**AWS CLI**
**Come elencare i flussi di dati**
L’esempio `list-streams` seguente elenca tutti i flussi di dati attivi nell’account e nella Regione correnti.
```
aws kinesis list-streams
```
Output:
```
{
"StreamNames": [
"samplestream",
"samplestream1"
]
}
```
Per ulteriori informazioni, consulta [Elenco di flussi](https://docs.aws.amazon.com/streams/latest/dev/kinesis-using-sdk-java-list-streams.html) nella *Guida per gli sviluppatori di Flusso di dati Amazon Kinesis*.
+ Per i dettagli sull'API, consulta [ListStreams AWS CLI](https://awscli.amazonaws.com/v2/documentation/api/latest/reference/kinesis/list-streams.html)*Command 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/kinesis#code-examples).
```
async fn show_streams(client: &Client) -> Result<(), Error> {
let resp = client.list_streams().send().await?;
println!("Stream names:");
let streams = resp.stream_names;
for stream in &streams {
println!(" {}", stream);
}
println!("Found {} stream(s)", streams.len());
Ok(())
}
```
+ Per i dettagli sulle API, consulta il *riferimento [ListStreams](https://docs.rs/aws-sdk-kinesis/latest/aws_sdk_kinesis/client/struct.Client.html#method.list_streams)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/kns#code-examples).
```
TRY.
oo_result = lo_kns->liststreams( " oo_result is returned for testing purposes. "
"Set Limit to specify that a maximum of streams should be returned."
iv_limit = iv_limit ).
DATA(lt_streams) = oo_result->get_streamnames( ).
MESSAGE 'Streams listed.' TYPE 'I'.
CATCH /aws1/cx_knslimitexceededex.
MESSAGE 'The request processing has failed because of a limit exceed exception.' TYPE 'E'.
ENDTRY.
```
+ Per i dettagli sulle API, [ListStreams](https://docs.aws.amazon.com/sdk-for-sap-abap/v1/api/latest/index.html)consulta *AWS SDK for SAP ABAP* API reference.
------
# Utilizzo `ListTagsForStream` con un AWS SDK o una CLI
Gli esempi di codice seguenti mostrano come utilizzare `ListTagsForStream`.
------
#### [ .NET ]
**SDK per .NET**
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/dotnetv3/Kinesis#code-examples).
```
using System;
using System.Collections.Generic;
using System.Threading.Tasks;
using Amazon.Kinesis;
using Amazon.Kinesis.Model;
///
/// Shows how to list the tags that have been attached to an Amazon Kinesis
/// stream.
///
public class ListTags
{
public static async Task Main()
{
IAmazonKinesis client = new AmazonKinesisClient();
string streamName = "AmazonKinesisStream";
await ListTagsAsync(client, streamName);
}
///
/// List the tags attached to a Kinesis stream.
///
/// An initialized Kinesis client object.
/// The name of the Kinesis stream for which you
/// wish to display tags.
public static async Task ListTagsAsync(IAmazonKinesis client, string streamName)
{
var request = new ListTagsForStreamRequest
{
StreamName = streamName,
Limit = 10,
};
var response = await client.ListTagsForStreamAsync(request);
DisplayTags(response.Tags);
while (response.HasMoreTags)
{
request.ExclusiveStartTagKey = response.Tags[response.Tags.Count - 1].Key;
response = await client.ListTagsForStreamAsync(request);
}
}
///
/// Displays the items in a list of Kinesis tags.
///
/// A list of the Tag objects to be displayed.
public static void DisplayTags(List tags)
{
tags
.ForEach(t => Console.WriteLine($"Key: {t.Key} Value: {t.Value}"));
}
}
```
+ Per i dettagli sull'API, [ListTagsForStream](https://docs.aws.amazon.com/goto/DotNetSDKV3/kinesis-2013-12-02/ListTagsForStream)consulta *AWS SDK per .NET API Reference*.
------
#### [ CLI ]
**AWS CLI**
**Come elencare i tag di un flusso di dati**
L’esempio `list-tags-for-stream` seguente elenca i tag associati al flusso di dati specificato.
```
aws kinesis list-tags-for-stream \
--stream-name samplestream
```
Output:
```
{
"Tags": [
{
"Key": "samplekey",
"Value": "example"
}
],
"HasMoreTags": false
}
```
Per ulteriori informazioni, consulta [Tagging dei flussi](https://docs.aws.amazon.com/streams/latest/dev/tagging.html) nella *Guida per gli sviluppatori di Flusso di dati Amazon Kinesis*.
+ Per i dettagli sull'API, consulta [ListTagsForStream AWS CLI](https://awscli.amazonaws.com/v2/documentation/api/latest/reference/kinesis/list-tags-for-stream.html)*Command Reference*.
------
# 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:
+ [Informazioni di base](kinesis_example_kinesis_Scenario_GettingStarted_section.md)
------
#### [ CLI ]
**AWS CLI**
**Come scrivere un record in un flusso di dati**
L’esempio `put-record` seguente scrive un unico record di dati nel flusso di dati specificato utilizzando la chiave di partizione specificata.
```
aws kinesis put-record \
--stream-name samplestream \
--data sampledatarecord \
--partition-key samplepartitionkey
```
Output:
```
{
"ShardId": "shardId-000000000009",
"SequenceNumber": "49600902273357540915989931256901506243878407835297513618",
"EncryptionType": "KMS"
}
```
Per ulteriori informazioni, consulta [Developing Producers Using the Amazon Kinesis Data Streams API AWS with the SDK for Java](https://docs.aws.amazon.com/streams/latest/dev/developing-producers-with-sdk.html) nella Amazon *Kinesis Data Streams* Developer Guide.
+ *Per i dettagli sull'API, consulta Command Reference. [PutRecord](https://awscli.amazonaws.com/v2/documentation/api/latest/reference/kinesis/put-record.html)AWS CLI *
------
#### [ 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/kinesis#code-examples).
```
import software.amazon.awssdk.core.SdkBytes;
import software.amazon.awssdk.regions.Region;
import software.amazon.awssdk.services.kinesis.KinesisClient;
import software.amazon.awssdk.services.kinesis.model.PutRecordRequest;
import software.amazon.awssdk.services.kinesis.model.KinesisException;
import software.amazon.awssdk.services.kinesis.model.DescribeStreamRequest;
import software.amazon.awssdk.services.kinesis.model.DescribeStreamResponse;
/**
* Before running this Java V2 code example, set up your development
* environment, including your credentials.
*
* For more information, see the following documentation topic:
*
* https://docs.aws.amazon.com/sdk-for-java/latest/developer-guide/get-started.html
*/
public class StockTradesWriter {
public static void main(String[] args) {
final String usage = """
Usage:
Where:
streamName - The Amazon Kinesis data stream to which records are written (for example, StockTradeStream)
""";
if (args.length != 1) {
System.out.println(usage);
System.exit(1);
}
String streamName = args[0];
Region region = Region.US_EAST_1;
KinesisClient kinesisClient = KinesisClient.builder()
.region(region)
.build();
// Ensure that the Kinesis Stream is valid.
validateStream(kinesisClient, streamName);
setStockData(kinesisClient, streamName);
kinesisClient.close();
}
public static void setStockData(KinesisClient kinesisClient, String streamName) {
try {
// Repeatedly send stock trades with a 100 milliseconds wait in between.
StockTradeGenerator stockTradeGenerator = new StockTradeGenerator();
// Put in 50 Records for this example.
int index = 50;
for (int x = 0; x < index; x++) {
StockTrade trade = stockTradeGenerator.getRandomTrade();
sendStockTrade(trade, kinesisClient, streamName);
Thread.sleep(100);
}
} catch (KinesisException | InterruptedException e) {
System.err.println(e.getMessage());
System.exit(1);
}
System.out.println("Done");
}
private static void sendStockTrade(StockTrade trade, KinesisClient kinesisClient,
String streamName) {
byte[] bytes = trade.toJsonAsBytes();
// The bytes could be null if there is an issue with the JSON serialization by
// the Jackson JSON library.
if (bytes == null) {
System.out.println("Could not get JSON bytes for stock trade");
return;
}
System.out.println("Putting trade: " + trade);
PutRecordRequest request = PutRecordRequest.builder()
.partitionKey(trade.getTickerSymbol()) // We use the ticker symbol as the partition key, explained in
// the Supplemental Information section below.
.streamName(streamName)
.data(SdkBytes.fromByteArray(bytes))
.build();
try {
kinesisClient.putRecord(request);
} catch (KinesisException e) {
System.err.println(e.getMessage());
}
}
private static void validateStream(KinesisClient kinesisClient, String streamName) {
try {
DescribeStreamRequest describeStreamRequest = DescribeStreamRequest.builder()
.streamName(streamName)
.build();
DescribeStreamResponse describeStreamResponse = kinesisClient.describeStream(describeStreamRequest);
if (!describeStreamResponse.streamDescription().streamStatus().toString().equals("ACTIVE")) {
System.err.println("Stream " + streamName + " is not active. Please wait a few moments and try again.");
System.exit(1);
}
} catch (KinesisException e) {
System.err.println("Error found while describing the stream " + streamName);
System.err.println(e);
System.exit(1);
}
}
}
```
+ Per i dettagli sull'API, [PutRecord](https://docs.aws.amazon.com/goto/SdkForJavaV2/kinesis-2013-12-02/PutRecord)consulta *AWS SDK for Java 2.x API Reference*.
------
#### [ PowerShell ]
**Strumenti per PowerShell V4**
**Esempio 1: scrive un record contenente la stringa fornita al parametro -Text.**
```
Write-KINRecord -Text "test data from string" -StreamName "mystream" -PartitionKey "Key1"
```
**Esempio 2: scrive un record contenente i dati contenuti nel file specificato. Il file viene trattato come una sequenza di byte. Pertanto, se contiene testo, deve essere scritto con la codifica necessaria prima di utilizzarlo con questo cmdlet.**
```
Write-KINRecord -FilePath "C:\TestData.txt" -StreamName "mystream" -PartitionKey "Key2"
```
+ Per i dettagli sull'API, vedere [PutRecord](https://docs.aws.amazon.com/powershell/v4/reference)in *AWS Strumenti per PowerShell Cmdlet Reference (*V4).
**Strumenti per V5 PowerShell **
**Esempio 1: scrive un record contenente la stringa fornita al parametro -Text.**
```
Write-KINRecord -Text "test data from string" -StreamName "mystream" -PartitionKey "Key1"
```
**Esempio 2: scrive un record contenente i dati contenuti nel file specificato. Il file viene trattato come una sequenza di byte. Pertanto, se contiene testo, deve essere scritto con la codifica necessaria prima di utilizzarlo con questo cmdlet.**
```
Write-KINRecord -FilePath "C:\TestData.txt" -StreamName "mystream" -PartitionKey "Key2"
```
+ Per i dettagli sull'API, vedere [PutRecord](https://docs.aws.amazon.com/powershell/v5/reference)in *AWS Strumenti per PowerShell Cmdlet Reference (*V5).
------
#### [ 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/kinesis#code-examples).
```
class KinesisStream:
"""Encapsulates a Kinesis stream."""
def __init__(self, kinesis_client):
"""
:param kinesis_client: A Boto3 Kinesis client.
"""
self.kinesis_client = kinesis_client
self.name = None
self.details = None
self.stream_exists_waiter = kinesis_client.get_waiter("stream_exists")
def put_record(self, data, partition_key):
"""
Puts data into the stream. The data is formatted as JSON before it is passed
to the stream.
:param data: The data to put in the stream.
:param partition_key: The partition key to use for the data.
:return: Metadata about the record, including its shard ID and sequence number.
"""
try:
response = self.kinesis_client.put_record(
StreamName=self.name, Data=json.dumps(data), PartitionKey=partition_key
)
logger.info("Put record in stream %s.", self.name)
except ClientError:
logger.exception("Couldn't put record in stream %s.", self.name)
raise
else:
return response
```
+ Per i dettagli sull'API, consulta [PutRecord AWS](https://docs.aws.amazon.com/goto/boto3/kinesis-2013-12-02/PutRecord)*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/kinesis#code-examples).
```
async fn add_record(client: &Client, stream: &str, key: &str, data: &str) -> Result<(), Error> {
let blob = Blob::new(data);
client
.put_record()
.data(blob)
.partition_key(key)
.stream_name(stream)
.send()
.await?;
println!("Put data into stream.");
Ok(())
}
```
+ Per i dettagli sulle API, consulta il *riferimento [PutRecord](https://docs.rs/aws-sdk-kinesis/latest/aws_sdk_kinesis/client/struct.Client.html#method.put_record)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/kns#code-examples).
```
TRY.
oo_result = lo_kns->putrecord( " oo_result is returned for testing purposes. "
iv_streamname = iv_stream_name
iv_data = iv_data
iv_partitionkey = iv_partition_key ).
MESSAGE 'Record created.' TYPE 'I'.
CATCH /aws1/cx_knsinvalidargumentex.
MESSAGE 'The specified argument was not valid.' TYPE 'E'.
CATCH /aws1/cx_knskmsaccessdeniedex.
MESSAGE 'You do not have permission to perform this AWS KMS action.' TYPE 'E'.
CATCH /aws1/cx_knskmsdisabledex.
MESSAGE 'KMS key used is disabled.' TYPE 'E'.
CATCH /aws1/cx_knskmsinvalidstateex.
MESSAGE 'KMS key used is in an invalid state. ' TYPE 'E'.
CATCH /aws1/cx_knskmsnotfoundex.
MESSAGE 'KMS key used is not found.' TYPE 'E'.
CATCH /aws1/cx_knskmsoptinrequired.
MESSAGE 'KMS key option is required.' TYPE 'E'.
CATCH /aws1/cx_knskmsthrottlingex.
MESSAGE 'The rate of requests to AWS KMS is exceeding the request quotas.' TYPE 'E'.
CATCH /aws1/cx_knsprovthruputexcdex.
MESSAGE 'The request rate for the stream is too high, or the requested data is too large for the available throughput.' TYPE 'E'.
CATCH /aws1/cx_knsresourcenotfoundex.
MESSAGE 'Resource being accessed is not found.' 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 `PutRecords` con un AWS SDK o una CLI
Gli esempi di codice seguenti mostrano come utilizzare `PutRecords`.
------
#### [ CLI ]
**AWS CLI**
**Come scrivere più record in un flusso di dati**
L’esempio `put-records` seguente scrive un record di dati utilizzando la chiave di partizione specificata e un altro record di dati utilizzando una chiave di partizione diversa in un’unica chiamata.
```
aws kinesis put-records \
--stream-name samplestream \
--records Data=blob1,PartitionKey=partitionkey1 Data=blob2,PartitionKey=partitionkey2
```
Output:
```
{
"FailedRecordCount": 0,
"Records": [
{
"SequenceNumber": "49600883331171471519674795588238531498465399900093808706",
"ShardId": "shardId-000000000004"
},
{
"SequenceNumber": "49600902273357540915989931256902715169698037101720764562",
"ShardId": "shardId-000000000009"
}
],
"EncryptionType": "KMS"
}
```
Per ulteriori informazioni, consulta [Developing Producers Using the Amazon Kinesis Data Streams API AWS with the SDK for Java](https://docs.aws.amazon.com/streams/latest/dev/developing-producers-with-sdk.html) nella Amazon *Kinesis Data Streams* Developer Guide.
+ *Per i dettagli sull'API, consulta Command Reference. [PutRecords](https://awscli.amazonaws.com/v2/documentation/api/latest/reference/kinesis/put-records.html)AWS CLI *
------
#### [ JavaScript ]
**SDK per JavaScript (v3)**
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/javascriptv3/example_code/kinesis#code-examples).
```
import { PutRecordsCommand, KinesisClient } from "@aws-sdk/client-kinesis";
/**
* Put multiple records into a Kinesis stream.
* @param {{ streamArn: string }} config
*/
export const main = async ({ streamArn }) => {
const client = new KinesisClient({});
try {
await client.send(
new PutRecordsCommand({
StreamARN: streamArn,
Records: [
{
Data: new Uint8Array(),
/**
* Determines which shard in the stream the data record is assigned to.
* Partition keys are Unicode strings with a maximum length limit of 256
* characters for each key. Amazon Kinesis Data Streams uses the partition
* key as input to a hash function that maps the partition key and
* associated data to a specific shard.
*/
PartitionKey: "TEST_KEY",
},
{
Data: new Uint8Array(),
PartitionKey: "TEST_KEY",
},
],
}),
);
} catch (caught) {
if (caught instanceof Error) {
//
} else {
throw caught;
}
}
};
// Call function if run directly.
import { fileURLToPath } from "node:url";
import { parseArgs } from "node:util";
if (process.argv[1] === fileURLToPath(import.meta.url)) {
const options = {
streamArn: {
type: "string",
description: "The ARN of the stream.",
},
};
const { values } = parseArgs({ options });
main(values);
}
```
+ Per i dettagli sull'API, [PutRecords](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/client/kinesis/command/PutRecordsCommand)consulta *AWS SDK per JavaScript API Reference*.
------
# Utilizzo `RegisterStreamConsumer` con un AWS SDK o una CLI
Gli esempi di codice seguenti mostrano come utilizzare `RegisterStreamConsumer`.
------
#### [ .NET ]
**SDK per .NET**
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/dotnetv3/Kinesis#code-examples).
```
using System;
using System.Threading.Tasks;
using Amazon.Kinesis;
using Amazon.Kinesis.Model;
///
/// This example shows how to register a consumer to an Amazon Kinesis
/// stream.
///
public class RegisterConsumer
{
public static async Task Main()
{
IAmazonKinesis client = new AmazonKinesisClient();
string consumerName = "NEW_CONSUMER_NAME";
string streamARN = "arn:aws:kinesis:us-east-2:000000000000:stream/AmazonKinesisStream";
var consumer = await RegisterConsumerAsync(client, consumerName, streamARN);
if (consumer is not null)
{
Console.WriteLine($"{consumer.ConsumerName}");
}
}
///
/// Registers the consumer to a Kinesis stream.
///
/// The initialized Kinesis client object.
/// A string representing the consumer.
/// The ARN of the stream.
/// A Consumer object that contains information about the consumer.
public static async Task RegisterConsumerAsync(IAmazonKinesis client, string consumerName, string streamARN)
{
var request = new RegisterStreamConsumerRequest
{
ConsumerName = consumerName,
StreamARN = streamARN,
};
var response = await client.RegisterStreamConsumerAsync(request);
return response.Consumer;
}
}
```
+ Per i dettagli sull'API, [RegisterStreamConsumer](https://docs.aws.amazon.com/goto/DotNetSDKV3/kinesis-2013-12-02/RegisterStreamConsumer)consulta *AWS SDK per .NET API Reference*.
------
#### [ CLI ]
**AWS CLI**
**Come registrare un consumer del flusso di dati**
L’esempio `register-stream-consumer` seguente registra un consumer denominato `KinesisConsumerApplication` con il flusso di dati specificato.
```
aws kinesis register-stream-consumer \
--stream-arn arn:aws:kinesis:us-west-2:012345678912:stream/samplestream \
--consumer-name KinesisConsumerApplication
```
Output:
```
{
"Consumer": {
"ConsumerName": "KinesisConsumerApplication",
"ConsumerARN": "arn:aws:kinesis:us-west-2: 123456789012:stream/samplestream/consumer/KinesisConsumerApplication:1572383852",
"ConsumerStatus": "CREATING",
"ConsumerCreationTimestamp": 1572383852.0
}
}
```
Per ulteriori informazioni, consulta [Sviluppo di applicazioni consumer con fan-out avanzato con l’API di Flusso dati Kinesis](https://docs.aws.amazon.com/streams/latest/dev/building-enhanced-consumers-api.html) nella *Guida per gli sviluppatori di Flusso di dati Amazon Kinesis*.
+ Per i dettagli sull'API, consulta [RegisterStreamConsumer AWS CLI](https://awscli.amazonaws.com/v2/documentation/api/latest/reference/kinesis/register-stream-consumer.html)*Command 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/kns#code-examples).
```
TRY.
oo_result = lo_kns->registerstreamconsumer( " oo_result is returned for testing purposes. "
iv_streamarn = iv_stream_arn
iv_consumername = iv_consumer_name ).
MESSAGE 'Stream consumer registered.' TYPE 'I'.
CATCH /aws1/cx_knsinvalidargumentex.
MESSAGE 'The specified argument was not valid.' TYPE 'E'.
CATCH /aws1/cx_sgmresourcelimitexcd.
MESSAGE 'You have reached the limit on the number of resources.' TYPE 'E'.
CATCH /aws1/cx_sgmresourceinuse.
MESSAGE 'Resource being accessed is in use.' TYPE 'E'.
CATCH /aws1/cx_sgmresourcenotfound.
MESSAGE 'Resource being accessed is not found.' TYPE 'E'.
ENDTRY.
```
+ Per i dettagli sulle API, [RegisterStreamConsumer](https://docs.aws.amazon.com/sdk-for-sap-abap/v1/api/latest/index.html)consulta *AWS SDK for SAP ABAP* API reference.
------
# Esempi serverless per Kinesis
I seguenti esempi di codice mostrano come usare Kinesis con. AWS SDKs
**Topics**
+ [Invocare una funzione Lambda da un trigger Kinesis](kinesis_example_serverless_Kinesis_Lambda_section.md)
+ [Segnalazione di errori di elementi batch per funzioni Lambda con un trigger Kinesis](kinesis_example_serverless_Kinesis_Lambda_batch_item_failures_section.md)
# Invocare una funzione Lambda da un trigger Kinesis
I seguenti esempi di codice spiegano come implementare una funzione Lambda che riceve un evento attivato dalla ricezione di record da un flusso Kinesis. La funzione recupera il payload Kinesis, lo decodifica da Base64 e registra il contenuto del record.
------
#### [ .NET ]
**SDK per .NET**
C'è altro da fare. GitHub Trova l’esempio completo e scopri come eseguire la configurazione e l’esecuzione nel repository di [Esempi serverless](https://github.com/aws-samples/serverless-snippets/tree/main/integration-kinesis-to-lambda).
Utilizzo di un evento Kinesis con Lambda tramite .NET.
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
using System.Text;
using Amazon.Lambda.Core;
using Amazon.Lambda.KinesisEvents;
using AWS.Lambda.Powertools.Logging;
// Assembly attribute to enable the Lambda function's JSON input to be converted into a .NET class.
[assembly: LambdaSerializer(typeof(Amazon.Lambda.Serialization.SystemTextJson.DefaultLambdaJsonSerializer))]
namespace KinesisIntegrationSampleCode;
public class Function
{
// Powertools Logger requires an environment variables against your function
// POWERTOOLS_SERVICE_NAME
[Logging(LogEvent = true)]
public async Task FunctionHandler(KinesisEvent evnt, ILambdaContext context)
{
if (evnt.Records.Count == 0)
{
Logger.LogInformation("Empty Kinesis Event received");
return;
}
foreach (var record in evnt.Records)
{
try
{
Logger.LogInformation($"Processed Event with EventId: {record.EventId}");
string data = await GetRecordDataAsync(record.Kinesis, context);
Logger.LogInformation($"Data: {data}");
// TODO: Do interesting work based on the new data
}
catch (Exception ex)
{
Logger.LogError($"An error occurred {ex.Message}");
throw;
}
}
Logger.LogInformation($"Successfully processed {evnt.Records.Count} records.");
}
private async Task GetRecordDataAsync(KinesisEvent.Record record, ILambdaContext context)
{
byte[] bytes = record.Data.ToArray();
string data = Encoding.UTF8.GetString(bytes);
await Task.CompletedTask; //Placeholder for actual async work
return data;
}
}
```
------
#### [ Go ]
**SDK per Go V2**
C'è dell'altro GitHub. Trova l’esempio completo e scopri come eseguire la configurazione e l’esecuzione nel repository di [Esempi serverless](https://github.com/aws-samples/serverless-snippets/tree/main/integration-kinesis-to-lambda).
Utilizzo di un evento Kinesis con Lambda tramite Go.
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
package main
import (
"context"
"log"
"github.com/aws/aws-lambda-go/events"
"github.com/aws/aws-lambda-go/lambda"
)
func handler(ctx context.Context, kinesisEvent events.KinesisEvent) error {
if len(kinesisEvent.Records) == 0 {
log.Printf("empty Kinesis event received")
return nil
}
for _, record := range kinesisEvent.Records {
log.Printf("processed Kinesis event with EventId: %v", record.EventID)
recordDataBytes := record.Kinesis.Data
recordDataText := string(recordDataBytes)
log.Printf("record data: %v", recordDataText)
// TODO: Do interesting work based on the new data
}
log.Printf("successfully processed %v records", len(kinesisEvent.Records))
return nil
}
func main() {
lambda.Start(handler)
}
```
------
#### [ Java ]
**SDK per Java 2.x**
C'è dell'altro GitHub. Trova l’esempio completo e scopri come eseguire la configurazione e l’esecuzione nel repository di [Esempi serverless](https://github.com/aws-samples/serverless-snippets/tree/main/integration-kinesis-to-lambda).
Utilizzo di un evento Kinesis con Lambda tramite Java.
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
package example;
import com.amazonaws.services.lambda.runtime.Context;
import com.amazonaws.services.lambda.runtime.LambdaLogger;
import com.amazonaws.services.lambda.runtime.RequestHandler;
import com.amazonaws.services.lambda.runtime.events.KinesisEvent;
public class Handler implements RequestHandler {
@Override
public Void handleRequest(final KinesisEvent event, final Context context) {
LambdaLogger logger = context.getLogger();
if (event.getRecords().isEmpty()) {
logger.log("Empty Kinesis Event received");
return null;
}
for (KinesisEvent.KinesisEventRecord record : event.getRecords()) {
try {
logger.log("Processed Event with EventId: "+record.getEventID());
String data = new String(record.getKinesis().getData().array());
logger.log("Data:"+ data);
// TODO: Do interesting work based on the new data
}
catch (Exception ex) {
logger.log("An error occurred:"+ex.getMessage());
throw ex;
}
}
logger.log("Successfully processed:"+event.getRecords().size()+" records");
return null;
}
}
```
------
#### [ JavaScript ]
**SDK per JavaScript (v3)**
C'è altro da fare. GitHub Trova l’esempio completo e scopri come eseguire la configurazione e l’esecuzione nel repository di [Esempi serverless](https://github.com/aws-samples/serverless-snippets/blob/main/integration-kinesis-to-lambda).
Consumo di un evento Kinesis con Lambda utilizzando. JavaScript
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
exports.handler = async (event, context) => {
for (const record of event.Records) {
try {
console.log(`Processed Kinesis Event - EventID: ${record.eventID}`);
const recordData = await getRecordDataAsync(record.kinesis);
console.log(`Record Data: ${recordData}`);
// TODO: Do interesting work based on the new data
} catch (err) {
console.error(`An error occurred ${err}`);
throw err;
}
}
console.log(`Successfully processed ${event.Records.length} records.`);
};
async function getRecordDataAsync(payload) {
var data = Buffer.from(payload.data, "base64").toString("utf-8");
await Promise.resolve(1); //Placeholder for actual async work
return data;
}
```
Consumo di un evento Kinesis con Lambda utilizzando. TypeScript
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
import {
KinesisStreamEvent,
Context,
KinesisStreamHandler,
KinesisStreamRecordPayload,
} from "aws-lambda";
import { Buffer } from "buffer";
import { Logger } from "@aws-lambda-powertools/logger";
const logger = new Logger({
logLevel: "INFO",
serviceName: "kinesis-stream-handler-sample",
});
export const functionHandler: KinesisStreamHandler = async (
event: KinesisStreamEvent,
context: Context
): Promise => {
for (const record of event.Records) {
try {
logger.info(`Processed Kinesis Event - EventID: ${record.eventID}`);
const recordData = await getRecordDataAsync(record.kinesis);
logger.info(`Record Data: ${recordData}`);
// TODO: Do interesting work based on the new data
} catch (err) {
logger.error(`An error occurred ${err}`);
throw err;
}
logger.info(`Successfully processed ${event.Records.length} records.`);
}
};
async function getRecordDataAsync(
payload: KinesisStreamRecordPayload
): Promise {
var data = Buffer.from(payload.data, "base64").toString("utf-8");
await Promise.resolve(1); //Placeholder for actual async work
return data;
}
```
------
#### [ PHP ]
**SDK per PHP**
C'è altro da fare. GitHub Trova l’esempio completo e scopri come eseguire la configurazione e l’esecuzione nel repository di [Esempi serverless](https://github.com/aws-samples/serverless-snippets/tree/main/integration-kinesis-to-lambda).
Utilizzo di un evento Kinesis con Lambda tramite PHP.
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
logger = $logger;
}
/**
* @throws JsonException
* @throws \Bref\Event\InvalidLambdaEvent
*/
public function handleKinesis(KinesisEvent $event, Context $context): void
{
$this->logger->info("Processing records");
$records = $event->getRecords();
foreach ($records as $record) {
$data = $record->getData();
$this->logger->info(json_encode($data));
// TODO: Do interesting work based on the new data
// Any exception thrown will be logged and the invocation will be marked as failed
}
$totalRecords = count($records);
$this->logger->info("Successfully processed $totalRecords records");
}
}
$logger = new StderrLogger();
return new Handler($logger);
```
------
#### [ Python ]
**SDK per Python (Boto3)**
C'è dell'altro GitHub. Trova l’esempio completo e scopri come eseguire la configurazione e l’esecuzione nel repository di [Esempi serverless](https://github.com/aws-samples/serverless-snippets/tree/main/integration-kinesis-to-lambda).
Utilizzo di un evento Kinesis con Lambda tramite Python.
```
# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
# SPDX-License-Identifier: Apache-2.0
import base64
def lambda_handler(event, context):
for record in event['Records']:
try:
print(f"Processed Kinesis Event - EventID: {record['eventID']}")
record_data = base64.b64decode(record['kinesis']['data']).decode('utf-8')
print(f"Record Data: {record_data}")
# TODO: Do interesting work based on the new data
except Exception as e:
print(f"An error occurred {e}")
raise e
print(f"Successfully processed {len(event['Records'])} records.")
```
------
#### [ Ruby ]
**SDK per Ruby**
C'è dell'altro GitHub. Trova l’esempio completo e scopri come eseguire la configurazione e l’esecuzione nel repository di [Esempi serverless](https://github.com/aws-samples/serverless-snippets/tree/main/integration-kinesis-to-lambda).
Utilizzo di un evento Kinesis con Lambda tramite Ruby.
```
# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
# SPDX-License-Identifier: Apache-2.0
require 'aws-sdk'
def lambda_handler(event:, context:)
event['Records'].each do |record|
begin
puts "Processed Kinesis Event - EventID: #{record['eventID']}"
record_data = get_record_data_async(record['kinesis'])
puts "Record Data: #{record_data}"
# TODO: Do interesting work based on the new data
rescue => err
$stderr.puts "An error occurred #{err}"
raise err
end
end
puts "Successfully processed #{event['Records'].length} records."
end
def get_record_data_async(payload)
data = Base64.decode64(payload['data']).force_encoding('UTF-8')
# Placeholder for actual async work
# You can use Ruby's asynchronous programming tools like async/await or fibers here.
return data
end
```
------
#### [ Rust ]
**SDK per Rust**
C'è altro da fare. GitHub Trova l’esempio completo e scopri come eseguire la configurazione e l’esecuzione nel repository di [Esempi serverless](https://github.com/aws-samples/serverless-snippets/tree/main/integration-kinesis-to-lambda).
Utilizzo di un evento Kinesis con Lambda tramite Rust.
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
use aws_lambda_events::event::kinesis::KinesisEvent;
use lambda_runtime::{run, service_fn, Error, LambdaEvent};
async fn function_handler(event: LambdaEvent) -> Result<(), Error> {
if event.payload.records.is_empty() {
tracing::info!("No records found. Exiting.");
return Ok(());
}
event.payload.records.iter().for_each(|record| {
tracing::info!("EventId: {}",record.event_id.as_deref().unwrap_or_default());
let record_data = std::str::from_utf8(&record.kinesis.data);
match record_data {
Ok(data) => {
// log the record data
tracing::info!("Data: {}", data);
}
Err(e) => {
tracing::error!("Error: {}", e);
}
}
});
tracing::info!(
"Successfully processed {} records",
event.payload.records.len()
);
Ok(())
}
#[tokio::main]
async fn main() -> Result<(), Error> {
tracing_subscriber::fmt()
.with_max_level(tracing::Level::INFO)
// disable printing the name of the module in every log line.
.with_target(false)
// disabling time is handy because CloudWatch will add the ingestion time.
.without_time()
.init();
run(service_fn(function_handler)).await
}
```
------
# Segnalazione di errori di elementi batch per funzioni Lambda con un trigger Kinesis
I seguenti esempi di codice spiegano come implementare una risposta batch parziale per funzioni Lambda che ricevono eventi da un flusso Kinesis. La funzione riporta gli errori degli elementi batch nella risposta, segnalando a Lambda di riprovare tali messaggi in un secondo momento.
------
#### [ .NET ]
**SDK per .NET**
C'è altro da fare. GitHub Trova l’esempio completo e scopri come eseguire la configurazione e l’esecuzione nel repository di [Esempi serverless](https://github.com/aws-samples/serverless-snippets/tree/main/integration-kinesis-to-lambda-with-batch-item-handling).
Segnalazione di errori di elementi batch di Kinesis con Lambda tramite .NET.
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
using System.Text;
using System.Text.Json.Serialization;
using Amazon.Lambda.Core;
using Amazon.Lambda.KinesisEvents;
using AWS.Lambda.Powertools.Logging;
// Assembly attribute to enable the Lambda function's JSON input to be converted into a .NET class.
[assembly: LambdaSerializer(typeof(Amazon.Lambda.Serialization.SystemTextJson.DefaultLambdaJsonSerializer))]
namespace KinesisIntegration;
public class Function
{
// Powertools Logger requires an environment variables against your function
// POWERTOOLS_SERVICE_NAME
[Logging(LogEvent = true)]
public async Task FunctionHandler(KinesisEvent evnt, ILambdaContext context)
{
if (evnt.Records.Count == 0)
{
Logger.LogInformation("Empty Kinesis Event received");
return new StreamsEventResponse();
}
foreach (var record in evnt.Records)
{
try
{
Logger.LogInformation($"Processed Event with EventId: {record.EventId}");
string data = await GetRecordDataAsync(record.Kinesis, context);
Logger.LogInformation($"Data: {data}");
// TODO: Do interesting work based on the new data
}
catch (Exception ex)
{
Logger.LogError($"An error occurred {ex.Message}");
/* Since we are working with streams, we can return the failed item immediately.
Lambda will immediately begin to retry processing from this failed item onwards. */
return new StreamsEventResponse
{
BatchItemFailures = new List
{
new StreamsEventResponse.BatchItemFailure { ItemIdentifier = record.Kinesis.SequenceNumber }
}
};
}
}
Logger.LogInformation($"Successfully processed {evnt.Records.Count} records.");
return new StreamsEventResponse();
}
private async Task GetRecordDataAsync(KinesisEvent.Record record, ILambdaContext context)
{
byte[] bytes = record.Data.ToArray();
string data = Encoding.UTF8.GetString(bytes);
await Task.CompletedTask; //Placeholder for actual async work
return data;
}
}
public class StreamsEventResponse
{
[JsonPropertyName("batchItemFailures")]
public IList BatchItemFailures { get; set; }
public class BatchItemFailure
{
[JsonPropertyName("itemIdentifier")]
public string ItemIdentifier { get; set; }
}
}
```
------
#### [ Go ]
**SDK per Go V2**
C'è dell'altro GitHub. Trova l’esempio completo e scopri come eseguire la configurazione e l’esecuzione nel repository di [Esempi serverless](https://github.com/aws-samples/serverless-snippets/tree/main/integration-kinesis-to-lambda-with-batch-item-handling).
Segnalazione di errori di elementi batch di Kinesis con Lambda tramite Go.
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
package main
import (
"context"
"fmt"
"github.com/aws/aws-lambda-go/events"
"github.com/aws/aws-lambda-go/lambda"
)
func handler(ctx context.Context, kinesisEvent events.KinesisEvent) (map[string]interface{}, error) {
batchItemFailures := []map[string]interface{}{}
for _, record := range kinesisEvent.Records {
curRecordSequenceNumber := ""
// Process your record
if /* Your record processing condition here */ {
curRecordSequenceNumber = record.Kinesis.SequenceNumber
}
// Add a condition to check if the record processing failed
if curRecordSequenceNumber != "" {
batchItemFailures = append(batchItemFailures, map[string]interface{}{"itemIdentifier": curRecordSequenceNumber})
}
}
kinesisBatchResponse := map[string]interface{}{
"batchItemFailures": batchItemFailures,
}
return kinesisBatchResponse, nil
}
func main() {
lambda.Start(handler)
}
```
------
#### [ Java ]
**SDK per Java 2.x**
C'è dell'altro GitHub. Trova l’esempio completo e scopri come eseguire la configurazione e l’esecuzione nel repository di [Esempi serverless](https://github.com/aws-samples/serverless-snippets/tree/main/integration-kinesis-to-lambda-with-batch-item-handling).
Segnalazione di errori di elementi batch di Kinesis con Lambda tramite Java.
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
import com.amazonaws.services.lambda.runtime.Context;
import com.amazonaws.services.lambda.runtime.RequestHandler;
import com.amazonaws.services.lambda.runtime.events.KinesisEvent;
import com.amazonaws.services.lambda.runtime.events.StreamsEventResponse;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.List;
public class ProcessKinesisRecords implements RequestHandler {
@Override
public StreamsEventResponse handleRequest(KinesisEvent input, Context context) {
List batchItemFailures = new ArrayList<>();
String curRecordSequenceNumber = "";
for (KinesisEvent.KinesisEventRecord kinesisEventRecord : input.getRecords()) {
try {
//Process your record
KinesisEvent.Record kinesisRecord = kinesisEventRecord.getKinesis();
curRecordSequenceNumber = kinesisRecord.getSequenceNumber();
} catch (Exception e) {
/* Since we are working with streams, we can return the failed item immediately.
Lambda will immediately begin to retry processing from this failed item onwards. */
batchItemFailures.add(new StreamsEventResponse.BatchItemFailure(curRecordSequenceNumber));
return new StreamsEventResponse(batchItemFailures);
}
}
return new StreamsEventResponse(batchItemFailures);
}
}
```
------
#### [ JavaScript ]
**SDK per JavaScript (v3)**
C'è altro da fare. GitHub Trova l’esempio completo e scopri come eseguire la configurazione e l’esecuzione nel repository di [Esempi serverless](https://github.com/aws-samples/serverless-snippets/blob/main/integration-kinesis-to-lambda-with-batch-item-handling).
Segnalazione di errori di elementi batch di Kinesis con Lambda tramite Javascript.
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
exports.handler = async (event, context) => {
for (const record of event.Records) {
try {
console.log(`Processed Kinesis Event - EventID: ${record.eventID}`);
const recordData = await getRecordDataAsync(record.kinesis);
console.log(`Record Data: ${recordData}`);
// TODO: Do interesting work based on the new data
} catch (err) {
console.error(`An error occurred ${err}`);
/* Since we are working with streams, we can return the failed item immediately.
Lambda will immediately begin to retry processing from this failed item onwards. */
return {
batchItemFailures: [{ itemIdentifier: record.kinesis.sequenceNumber }],
};
}
}
console.log(`Successfully processed ${event.Records.length} records.`);
return { batchItemFailures: [] };
};
async function getRecordDataAsync(payload) {
var data = Buffer.from(payload.data, "base64").toString("utf-8");
await Promise.resolve(1); //Placeholder for actual async work
return data;
}
```
Segnalazione degli errori degli elementi batch di Kinesis utilizzando Lambda. TypeScript
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
import {
KinesisStreamEvent,
Context,
KinesisStreamHandler,
KinesisStreamRecordPayload,
KinesisStreamBatchResponse,
} from "aws-lambda";
import { Buffer } from "buffer";
import { Logger } from "@aws-lambda-powertools/logger";
const logger = new Logger({
logLevel: "INFO",
serviceName: "kinesis-stream-handler-sample",
});
export const functionHandler: KinesisStreamHandler = async (
event: KinesisStreamEvent,
context: Context
): Promise => {
for (const record of event.Records) {
try {
logger.info(`Processed Kinesis Event - EventID: ${record.eventID}`);
const recordData = await getRecordDataAsync(record.kinesis);
logger.info(`Record Data: ${recordData}`);
// TODO: Do interesting work based on the new data
} catch (err) {
logger.error(`An error occurred ${err}`);
/* Since we are working with streams, we can return the failed item immediately.
Lambda will immediately begin to retry processing from this failed item onwards. */
return {
batchItemFailures: [{ itemIdentifier: record.kinesis.sequenceNumber }],
};
}
}
logger.info(`Successfully processed ${event.Records.length} records.`);
return { batchItemFailures: [] };
};
async function getRecordDataAsync(
payload: KinesisStreamRecordPayload
): Promise {
var data = Buffer.from(payload.data, "base64").toString("utf-8");
await Promise.resolve(1); //Placeholder for actual async work
return data;
}
```
------
#### [ PHP ]
**SDK per PHP**
C'è altro da fare. GitHub Trova l’esempio completo e scopri come eseguire la configurazione e l’esecuzione nel repository di [Esempi serverless](https://github.com/aws-samples/serverless-snippets/tree/main/integration-kinesis-to-lambda-with-batch-item-handling).
Segnalazione di errori di elementi batch di Kinesis con Lambda tramite PHP.
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
logger = $logger;
}
/**
* @throws JsonException
* @throws \Bref\Event\InvalidLambdaEvent
*/
public function handle(mixed $event, Context $context): array
{
$kinesisEvent = new KinesisEvent($event);
$this->logger->info("Processing records");
$records = $kinesisEvent->getRecords();
$failedRecords = [];
foreach ($records as $record) {
try {
$data = $record->getData();
$this->logger->info(json_encode($data));
// TODO: Do interesting work based on the new data
} catch (Exception $e) {
$this->logger->error($e->getMessage());
// failed processing the record
$failedRecords[] = $record->getSequenceNumber();
}
}
$totalRecords = count($records);
$this->logger->info("Successfully processed $totalRecords records");
// change format for the response
$failures = array_map(
fn(string $sequenceNumber) => ['itemIdentifier' => $sequenceNumber],
$failedRecords
);
return [
'batchItemFailures' => $failures
];
}
}
$logger = new StderrLogger();
return new Handler($logger);
```
------
#### [ Python ]
**SDK per Python (Boto3)**
C'è dell'altro GitHub. Trova l’esempio completo e scopri come eseguire la configurazione e l’esecuzione nel repository di [Esempi serverless](https://github.com/aws-samples/serverless-snippets/tree/main/integration-kinesis-to-lambda-with-batch-item-handling).
Segnalazione di errori di elementi batch di Kinesis con Lambda tramite Python.
```
# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
# SPDX-License-Identifier: Apache-2.0
def handler(event, context):
records = event.get("Records")
curRecordSequenceNumber = ""
for record in records:
try:
# Process your record
curRecordSequenceNumber = record["kinesis"]["sequenceNumber"]
except Exception as e:
# Return failed record's sequence number
return {"batchItemFailures":[{"itemIdentifier": curRecordSequenceNumber}]}
return {"batchItemFailures":[]}
```
------
#### [ Ruby ]
**SDK per Ruby**
C'è dell'altro GitHub. Trova l’esempio completo e scopri come eseguire la configurazione e l’esecuzione nel repository di [Esempi serverless](https://github.com/aws-samples/serverless-snippets/tree/main/integration-kinesis-to-lambda-with-batch-item-handling).
Segnalazione di errori di elementi batch di Kinesis con Lambda tramite Ruby.
```
# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
# SPDX-License-Identifier: Apache-2.0
require 'aws-sdk'
def lambda_handler(event:, context:)
batch_item_failures = []
event['Records'].each do |record|
begin
puts "Processed Kinesis Event - EventID: #{record['eventID']}"
record_data = get_record_data_async(record['kinesis'])
puts "Record Data: #{record_data}"
# TODO: Do interesting work based on the new data
rescue StandardError => err
puts "An error occurred #{err}"
# Since we are working with streams, we can return the failed item immediately.
# Lambda will immediately begin to retry processing from this failed item onwards.
return { batchItemFailures: [{ itemIdentifier: record['kinesis']['sequenceNumber'] }] }
end
end
puts "Successfully processed #{event['Records'].length} records."
{ batchItemFailures: batch_item_failures }
end
def get_record_data_async(payload)
data = Base64.decode64(payload['data']).force_encoding('utf-8')
# Placeholder for actual async work
sleep(1)
data
end
```
------
#### [ Rust ]
**SDK per Rust**
C'è altro da fare. GitHub Trova l’esempio completo e scopri come eseguire la configurazione e l’esecuzione nel repository di [Esempi serverless](https://github.com/aws-samples/serverless-snippets/tree/main/integration-kinesis-to-lambda-with-batch-item-handling).
Segnalazione di errori di elementi batch di Kinesis con Lambda tramite Rust.
```
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
use aws_lambda_events::{
event::kinesis::KinesisEvent,
kinesis::KinesisEventRecord,
streams::{KinesisBatchItemFailure, KinesisEventResponse},
};
use lambda_runtime::{run, service_fn, Error, LambdaEvent};
async fn function_handler(event: LambdaEvent) -> Result {
let mut response = KinesisEventResponse {
batch_item_failures: vec![],
};
if event.payload.records.is_empty() {
tracing::info!("No records found. Exiting.");
return Ok(response);
}
for record in &event.payload.records {
tracing::info!(
"EventId: {}",
record.event_id.as_deref().unwrap_or_default()
);
let record_processing_result = process_record(record);
if record_processing_result.is_err() {
response.batch_item_failures.push(KinesisBatchItemFailure {
item_identifier: record.kinesis.sequence_number.clone(),
});
/* Since we are working with streams, we can return the failed item immediately.
Lambda will immediately begin to retry processing from this failed item onwards. */
return Ok(response);
}
}
tracing::info!(
"Successfully processed {} records",
event.payload.records.len()
);
Ok(response)
}
fn process_record(record: &KinesisEventRecord) -> Result<(), Error> {
let record_data = std::str::from_utf8(record.kinesis.data.as_slice());
if let Some(err) = record_data.err() {
tracing::error!("Error: {}", err);
return Err(Error::from(err));
}
let record_data = record_data.unwrap_or_default();
// do something interesting with the data
tracing::info!("Data: {}", record_data);
Ok(())
}
#[tokio::main]
async fn main() -> Result<(), Error> {
tracing_subscriber::fmt()
.with_max_level(tracing::Level::INFO)
// disable printing the name of the module in every log line.
.with_target(false)
// disabling time is handy because CloudWatch will add the ingestion time.
.without_time()
.init();
run(service_fn(function_handler)).await
}
```
------