Die vorliegende Übersetzung wurde maschinell erstellt. Im Falle eines Konflikts oder eines Widerspruchs zwischen dieser übersetzten Fassung und der englischen Fassung (einschließlich infolge von Verzögerungen bei der Übersetzung) ist die englische Fassung maßgeblich.
Verwenden von StreamManagerClient für die Arbeit mit Streams
Benutzerdefinierte Greengrass-Komponenten, die auf dem Greengrass-Kerngerät ausgeführt werden, können das -StreamManagerClient
Objekt im Stream Manager SDK verwenden, um Streams im Stream Manager zu erstellen und dann mit den Streams zu interagieren. Wenn eine Komponente einen Stream erstellt, definiert sie die AWS Cloud Ziele, die Priorisierung und andere Export- und Datenaufbewahrungsrichtlinien für den Stream. Um Daten an den Stream-Manager zu senden, hängen Komponenten die Daten an den Stream an. Wenn ein Exportziel für den Stream definiert ist, exportiert Stream Manager den Stream automatisch.
In der Regel sind Clients von Stream Manager benutzerdefinierte Greengrass-Komponenten. Wenn Ihr Geschäftsfall dies erfordert, können Sie auch zulassen, dass Nichtkomponentenprozesse, die auf dem Greengrass-Kern (z. B. einem Docker-Container) ausgeführt werden, mit dem Stream-Manager interagieren. Weitere Informationen finden Sie unter Client-Authentifizierung.
Die Ausschnitte in diesem Thema zeigen Ihnen, wie Clients StreamManagerClient
Methoden aufrufen, um mit Streams zu arbeiten. Für Implementierungsdetails zu den Methoden und ihren Argumenten verwenden Sie die Links zur SDK-Referenz, die nach jedem Ausschnitt aufgeführt ist.
Wenn Sie Stream Manager in einer Lambda-Funktion verwenden, sollte Ihre Lambda-Funktion StreamManagerClient
außerhalb des Funktionshandlers instanziieren. Wenn sie in dem Handler instanziiert wird, erstellt die Funktion bei jedem Aufruf eine client
und eine Verbindung zum Stream-Manager.
Wenn Sie StreamManagerClient
in dem Handler instanziieren, müssen Sie die close()
-Methode explizit aufrufen, wenn die client
seine Arbeit abschließt. Andernfalls hält der client
die Verbindung offen, und ein anderer Thread läuft, bis das Skript beendet wird.
StreamManagerClient
unterstützt die folgenden Operationen:
Erstellen eines Nachrichten-Streams
Um einen Stream zu erstellen, ruft eine benutzerdefinierte Greengrass-Komponente die Erstellungsmethode auf und übergibt ein -MessageStreamDefinition
Objekt. Dieses Objekt gibt den eindeutigen Namen für den Stream an und definiert, wie der Stream-Manager neue Daten verarbeiten soll, wenn die maximale Stream-Größe erreicht ist. Sie können mit MessageStreamDefinition
seinen Datentypen (z. B. ExportDefinition
, StrategyOnFull
, und Persistence
) andere Stream-Eigenschaften definieren. Dazu zählen:
-
Die Zielziele AWS IoT Analytics, Kinesis Data StreamsAWS IoT SiteWise, und Amazon S3 für automatische Exporte. Weitere Informationen finden Sie unter Exportkonfigurationen für unterstützte AWS Cloud Ziele.
-
Export-Priorität. Stream-Manager exportiert Streams mit höherer Priorität vor Streams mit niedrigerer Priorität.
-
Maximale Batchgröße und Batch-Intervall für AWS IoT Analytics, Kinesis Data Streams und AWS IoT SiteWise Ziele. Der Stream-Manager exportiert Nachrichten, wenn eine der Bedingungen erfüllt ist.
-
T ime-to-live (TTL). Die Zeitspanne, um sicherzustellen, dass die Streamdaten für die Verarbeitung verfügbar sind. Sie sollten sicherstellen, dass die Daten innerhalb dieses Zeitraums verbraucht werden können. Dies ist keine Löschrichtlinie. Die Daten werden möglicherweise nicht unmittelbar nach dem TTL-Zeitraum gelöscht.
-
Streampersistenz. Wählen Sie, ob Streams im Dateisystem gespeichert werden sollen, um Daten über Core-Neustarts hinweg zu speichern oder Streams im Speicher zu speichern.
-
Startsequenznummer. Geben Sie die Sequenznummer der Nachricht an, die als Startnachricht im Export verwendet werden soll.
Weitere Informationen zu finden Sie MessageStreamDefinition
in der SDK-Referenz für Ihre Zielsprache:
StreamManagerClient
bietet auch ein Ziel, mit dem Sie Streams auf einen HTTP-Server exportieren können. Dieses Ziel dient nur zu Testzwecken. Es ist nicht stabil oder wird nicht für die Verwendung in Produktionsumgebungen unterstützt.
Nachdem ein Stream erstellt wurde, können Ihre Greengrass-Komponenten Nachrichten an den Stream anhängen, um Daten für den Export zu senden und Nachrichten aus dem Stream zur lokalen Verarbeitung zu lesen. Die Anzahl der Streams, die Sie erstellen, hängt von Ihren Hardwarefunktionen und Ihrem Geschäftsfall ab. Eine Strategie besteht darin, einen Stream für jeden Zielkanal in AWS IoT Analytics oder Kinesis Data Stream zu erstellen, Sie können jedoch mehrere Ziele für einen Stream definieren. Ein Stream hat eine dauerhafte Lebensdauer.
Voraussetzungen
Für diesen Vorgang gelten die folgenden Anforderungen:
Beispiele
Das folgende Snippet erstellt einen Stream mit dem Namen StreamName
. Es definiert Stream-Eigenschaften in den untergeordneten Datentypen MessageStreamDefinition
und .
- Python
-
client = StreamManagerClient()
try:
client.create_message_stream(MessageStreamDefinition(
name="StreamName", # Required.
max_size=268435456, # Default is 256 MB.
stream_segment_size=16777216, # Default is 16 MB.
time_to_live_millis=None, # By default, no TTL is enabled.
strategy_on_full=StrategyOnFull.OverwriteOldestData, # Required.
persistence=Persistence.File, # Default is File.
flush_on_write=False, # Default is false.
export_definition=ExportDefinition( # Optional. Choose where/how the stream is exported to the AWS Cloud.
kinesis=None,
iot_analytics=None,
iot_sitewise=None,
s3_task_executor=None
)
))
except StreamManagerException:
pass
# Properly handle errors.
except ConnectionError or asyncio.TimeoutError:
pass
# Properly handle errors.
Python-SDK-Referenz: create_message_stream | MessageStreamDefinition
- Java
-
try (final StreamManagerClient client = StreamManagerClientFactory.standard().build()) {
client.createMessageStream(
new MessageStreamDefinition()
.withName("StreamName") // Required.
.withMaxSize(268435456L) // Default is 256 MB.
.withStreamSegmentSize(16777216L) // Default is 16 MB.
.withTimeToLiveMillis(null) // By default, no TTL is enabled.
.withStrategyOnFull(StrategyOnFull.OverwriteOldestData) // Required.
.withPersistence(Persistence.File) // Default is File.
.withFlushOnWrite(false) // Default is false.
.withExportDefinition( // Optional. Choose where/how the stream is exported to the AWS Cloud.
new ExportDefinition()
.withKinesis(null)
.withIotAnalytics(null)
.withIotSitewise(null)
.withS3(null)
)
);
} catch (StreamManagerException e) {
// Properly handle exception.
}
Java-SDK-Referenz: createMessageStream | MessageStreamDefinition
- Node.js
-
const client = new StreamManagerClient();
client.onConnected(async () => {
try {
await client.createMessageStream(
new MessageStreamDefinition()
.withName("StreamName") // Required.
.withMaxSize(268435456) // Default is 256 MB.
.withStreamSegmentSize(16777216) // Default is 16 MB.
.withTimeToLiveMillis(null) // By default, no TTL is enabled.
.withStrategyOnFull(StrategyOnFull.OverwriteOldestData) // Required.
.withPersistence(Persistence.File) // Default is File.
.withFlushOnWrite(false) // Default is false.
.withExportDefinition( // Optional. Choose where/how the stream is exported to the AWS Cloud.
new ExportDefinition()
.withKinesis(null)
.withIotAnalytics(null)
.withIotSiteWise(null)
.withS3(null)
)
);
} catch (e) {
// Properly handle errors.
}
});
client.onError((err) => {
// Properly handle connection errors.
// This is called only when the connection to the StreamManager server fails.
});
Node.js SDK-Referenz: createMessageStream | MessageStreamDefinition
Weitere Informationen zum Konfigurieren von Exportzielen finden Sie unter Exportkonfigurationen für unterstützte AWS Cloud Ziele.
Anhängen einer Nachricht
Um Daten zum Export an den Stream-Manager zu senden, hängen Ihre Greengrass-Komponenten die Daten an den Ziel-Stream an. Das Exportziel bestimmt den Datentyp, der an diese Methode übergeben werden soll.
Voraussetzungen
Für diesen Vorgang gelten die folgenden Anforderungen:
Beispiele
AWS IoT Analytics oder Kinesis-Data-Streams-Exportziele
Das folgende Snippet fügt eine Nachricht an den Stream namens StreamName
an. Für Ziele von AWS IoT Analytics oder Kinesis Data Streams hängen Ihre Greengrass-Komponenten einen Datenblob an.
Für diesen Ausschnitt gelten die folgenden Anforderungen:
- Python
-
client = StreamManagerClient()
try:
sequence_number = client.append_message(stream_name="StreamName", data=b'Arbitrary bytes data')
except StreamManagerException:
pass
# Properly handle errors.
except ConnectionError or asyncio.TimeoutError:
pass
# Properly handle errors.
Python-SDK-Referenz: append_message
- Java
-
try (final StreamManagerClient client = StreamManagerClientFactory.standard().build()) {
long sequenceNumber = client.appendMessage("StreamName", "Arbitrary byte array".getBytes());
} catch (StreamManagerException e) {
// Properly handle exception.
}
Java-SDK-Referenz: appendMessage
- Node.js
-
const client = new StreamManagerClient();
client.onConnected(async () => {
try {
const sequenceNumber = await client.appendMessage("StreamName", Buffer.from("Arbitrary byte array"));
} catch (e) {
// Properly handle errors.
}
});
client.onError((err) => {
// Properly handle connection errors.
// This is called only when the connection to the StreamManager server fails.
});
Node.js SDK-Referenz: appendMessage
AWS IoT SiteWise Exportziele
Das folgende Snippet fügt eine Nachricht an den Stream namens StreamName
an. Für AWS IoT SiteWise Ziele hängen Ihre Greengrass-Komponenten ein serialisiertes PutAssetPropertyValueEntry
Objekt an. Weitere Informationen finden Sie unter Exportieren in AWS IoT SiteWise.
Wenn Sie Daten an sendenAWS IoT SiteWise, müssen Ihre Daten die Anforderungen der BatchPutAssetPropertyValue
Aktion erfüllen. Weitere Informationen finden Sie unter BatchPutAssetPropertyValue in der AWS IoT SiteWise-API-Referenz.
Für diesen Ausschnitt gelten die folgenden Anforderungen:
- Python
-
client = StreamManagerClient()
try:
# SiteWise requires unique timestamps in all messages and also needs timestamps not earlier
# than 10 minutes in the past. Add some randomness to time and offset.
# Note: To create a new asset property data, you should use the classes defined in the
# greengrasssdk.stream_manager module.
time_in_nanos = TimeInNanos(
time_in_seconds=calendar.timegm(time.gmtime()) - random.randint(0, 60), offset_in_nanos=random.randint(0, 10000)
)
variant = Variant(double_value=random.random())
asset = [AssetPropertyValue(value=variant, quality=Quality.GOOD, timestamp=time_in_nanos)]
putAssetPropertyValueEntry = PutAssetPropertyValueEntry(entry_id=str(uuid.uuid4()), property_alias="PropertyAlias", property_values=asset)
sequence_number = client.append_message(stream_name="StreamName", Util.validate_and_serialize_to_json_bytes(putAssetPropertyValueEntry))
except StreamManagerException:
pass
# Properly handle errors.
except ConnectionError or asyncio.TimeoutError:
pass
# Properly handle errors.
Python-SDK-Referenz: append_message | PutAssetPropertyValueEntry
- Java
-
try (final StreamManagerClient client = GreengrassClientBuilder.streamManagerClient().build()) {
Random rand = new Random();
// Note: To create a new asset property data, you should use the classes defined in the
// com.amazonaws.greengrass.streammanager.model.sitewise package.
List<AssetPropertyValue> entries = new ArrayList<>() ;
// IoTSiteWise requires unique timestamps in all messages and also needs timestamps not earlier
// than 10 minutes in the past. Add some randomness to time and offset.
final int maxTimeRandomness = 60;
final int maxOffsetRandomness = 10000;
double randomValue = rand.nextDouble();
TimeInNanos timestamp = new TimeInNanos()
.withTimeInSeconds(Instant.now().getEpochSecond() - rand.nextInt(maxTimeRandomness))
.withOffsetInNanos((long) (rand.nextInt(maxOffsetRandomness)));
AssetPropertyValue entry = new AssetPropertyValue()
.withValue(new Variant().withDoubleValue(randomValue))
.withQuality(Quality.GOOD)
.withTimestamp(timestamp);
entries.add(entry);
PutAssetPropertyValueEntry putAssetPropertyValueEntry = new PutAssetPropertyValueEntry()
.withEntryId(UUID.randomUUID().toString())
.withPropertyAlias("PropertyAlias")
.withPropertyValues(entries);
long sequenceNumber = client.appendMessage("StreamName", ValidateAndSerialize.validateAndSerializeToJsonBytes(putAssetPropertyValueEntry));
} catch (StreamManagerException e) {
// Properly handle exception.
}
Java-SDK-Referenz: appendMessage | PutAssetPropertyValueEntry
- Node.js
-
const client = new StreamManagerClient();
client.onConnected(async () => {
try {
const maxTimeRandomness = 60;
const maxOffsetRandomness = 10000;
const randomValue = Math.random();
// Note: To create a new asset property data, you should use the classes defined in the
// aws-greengrass-core-sdk StreamManager module.
const timestamp = new TimeInNanos()
.withTimeInSeconds(Math.round(Date.now() / 1000) - Math.floor(Math.random() - maxTimeRandomness))
.withOffsetInNanos(Math.floor(Math.random() * maxOffsetRandomness));
const entry = new AssetPropertyValue()
.withValue(new Variant().withDoubleValue(randomValue))
.withQuality(Quality.GOOD)
.withTimestamp(timestamp);
const putAssetPropertyValueEntry = new PutAssetPropertyValueEntry()
.withEntryId(`${ENTRY_ID_PREFIX}${i}`)
.withPropertyAlias("PropertyAlias")
.withPropertyValues([entry]);
const sequenceNumber = await client.appendMessage("StreamName", util.validateAndSerializeToJsonBytes(putAssetPropertyValueEntry));
} catch (e) {
// Properly handle errors.
}
});
client.onError((err) => {
// Properly handle connection errors.
// This is called only when the connection to the StreamManager server fails.
});
Node.js SDK-Referenz: appendMessage | PutAssetPropertyValueEntry
Amazon S3-Exportziele
Der folgende Codeausschnitt hängt eine Exportaufgabe an den Stream mit dem Namen anStreamName
. Für Amazon S3-Ziele hängen Ihre Greengrass-Komponenten ein serialisiertes S3ExportTaskDefinition
Objekt an, das Informationen über die Quelleingabedatei und das Amazon S3-Zielobjekt enthält. Wenn das angegebene Objekt nicht vorhanden ist, erstellt Stream Manager es für Sie. Weitere Informationen finden Sie unter Exportieren nach Amazon S3.
Für diesen Ausschnitt gelten die folgenden Anforderungen:
- Python
-
client = StreamManagerClient()
try:
# Append an Amazon S3 Task definition and print the sequence number.
s3_export_task_definition = S3ExportTaskDefinition(input_url="URLToFile", bucket="BucketName", key="KeyName")
sequence_number = client.append_message(stream_name="StreamName", Util.validate_and_serialize_to_json_bytes(s3_export_task_definition))
except StreamManagerException:
pass
# Properly handle errors.
except ConnectionError or asyncio.TimeoutError:
pass
# Properly handle errors.
Python-SDK-Referenz: append_message | S3ExportTaskDefinition
- Java
-
try (final StreamManagerClient client = GreengrassClientBuilder.streamManagerClient().build()) {
// Append an Amazon S3 export task definition and print the sequence number.
S3ExportTaskDefinition s3ExportTaskDefinition = new S3ExportTaskDefinition()
.withBucket("BucketName")
.withKey("KeyName")
.withInputUrl("URLToFile");
long sequenceNumber = client.appendMessage("StreamName", ValidateAndSerialize.validateAndSerializeToJsonBytes(s3ExportTaskDefinition));
} catch (StreamManagerException e) {
// Properly handle exception.
}
Java-SDK-Referenz: appendMessage | S3ExportTaskDefinition
- Node.js
-
const client = new StreamManagerClient();
client.onConnected(async () => {
try {
// Append an Amazon S3 export task definition and print the sequence number.
const taskDefinition = new S3ExportTaskDefinition()
.withBucket("BucketName")
.withKey("KeyName")
.withInputUrl("URLToFile");
const sequenceNumber = await client.appendMessage("StreamName", util.validateAndSerializeToJsonBytes(taskDefinition)));
} catch (e) {
// Properly handle errors.
}
});
client.onError((err) => {
// Properly handle connection errors.
// This is called only when the connection to the StreamManager server fails.
});
Node.js SDK-Referenz: appendMessage | S3ExportTaskDefinition
Lesen von Nachrichten
Lesen von Nachrichten aus einem Stream.
Voraussetzungen
Für diesen Vorgang gelten die folgenden Anforderungen:
Beispiele
Das folgende Snippet liest Nachrichten aus dem Stream namens StreamName
. Die Read-Methode verwendet ein optionales ReadMessagesOptions
-Objekt, das die Sequenznummer angibt, von der aus mit dem Lesen begonnen werden soll, die minimale und maximale Anzahl zu lesen und ein Timeout für das Lesen von Nachrichten.
- Python
-
client = StreamManagerClient()
try:
message_list = client.read_messages(
stream_name="StreamName",
# By default, if no options are specified, it tries to read one message from the beginning of the stream.
options=ReadMessagesOptions(
desired_start_sequence_number=100,
# Try to read from sequence number 100 or greater. By default, this is 0.
min_message_count=10,
# Try to read 10 messages. If 10 messages are not available, then NotEnoughMessagesException is raised. By default, this is 1.
max_message_count=100, # Accept up to 100 messages. By default this is 1.
read_timeout_millis=5000
# Try to wait at most 5 seconds for the min_messsage_count to be fulfilled. By default, this is 0, which immediately returns the messages or an exception.
)
)
except StreamManagerException:
pass
# Properly handle errors.
except ConnectionError or asyncio.TimeoutError:
pass
# Properly handle errors.
Python-SDK-Referenz: read_messages | ReadMessagesOptions
- Java
-
try (final StreamManagerClient client = StreamManagerClientFactory.standard().build()) {
List<Message> messages = client.readMessages("StreamName",
// By default, if no options are specified, it tries to read one message from the beginning of the stream.
new ReadMessagesOptions()
// Try to read from sequence number 100 or greater. By default this is 0.
.withDesiredStartSequenceNumber(100L)
// Try to read 10 messages. If 10 messages are not available, then NotEnoughMessagesException is raised. By default, this is 1.
.withMinMessageCount(10L)
// Accept up to 100 messages. By default this is 1.
.withMaxMessageCount(100L)
// Try to wait at most 5 seconds for the min_messsage_count to be fulfilled. By default, this is 0, which immediately returns the messages or an exception.
.withReadTimeoutMillis(Duration.ofSeconds(5L).toMillis())
);
} catch (StreamManagerException e) {
// Properly handle exception.
}
Java-SDK-Referenz: readMessages | ReadMessagesOptions
- Node.js
-
const client = new StreamManagerClient();
client.onConnected(async () => {
try {
const messages = await client.readMessages("StreamName",
// By default, if no options are specified, it tries to read one message from the beginning of the stream.
new ReadMessagesOptions()
// Try to read from sequence number 100 or greater. By default this is 0.
.withDesiredStartSequenceNumber(100)
// Try to read 10 messages. If 10 messages are not available, then NotEnoughMessagesException is thrown. By default, this is 1.
.withMinMessageCount(10)
// Accept up to 100 messages. By default this is 1.
.withMaxMessageCount(100)
// Try to wait at most 5 seconds for the minMessageCount to be fulfilled. By default, this is 0, which immediately returns the messages or an exception.
.withReadTimeoutMillis(5 * 1000)
);
} catch (e) {
// Properly handle errors.
}
});
client.onError((err) => {
// Properly handle connection errors.
// This is called only when the connection to the StreamManager server fails.
});
Node.js SDK-Referenz: readMessages | ReadMessagesOptions
Auflisten von Streams
Rufen Sie die Liste der Streams im Stream-Manager ab.
Voraussetzungen
Für diesen Vorgang gelten die folgenden Anforderungen:
Beispiele
Das folgende Snippet ruft eine Liste der Streams (nach Namen) im Stream-Manager ab.
- Python
-
client = StreamManagerClient()
try:
stream_names = client.list_streams()
except StreamManagerException:
pass
# Properly handle errors.
except ConnectionError or asyncio.TimeoutError:
pass
# Properly handle errors.
Python-SDK-Referenz: list_streams
- Java
-
try (final StreamManagerClient client = StreamManagerClientFactory.standard().build()) {
List<String> streamNames = client.listStreams();
} catch (StreamManagerException e) {
// Properly handle exception.
}
Java-SDK-Referenz: listStreams
- Node.js
-
const client = new StreamManagerClient();
client.onConnected(async () => {
try {
const streams = await client.listStreams();
} catch (e) {
// Properly handle errors.
}
});
client.onError((err) => {
// Properly handle connection errors.
// This is called only when the connection to the StreamManager server fails.
});
Node.js SDK-Referenz: listStreams
Beschreiben eines Nachrichten-Streams
Abrufen von Metadaten zu einem Stream, einschließlich Stream-Definition, Größe und Exportstatus.
Voraussetzungen
Für diesen Vorgang gelten die folgenden Anforderungen:
Beispiele
Das folgende Snippet ruft Metadaten über den Stream mit dem Namen StreamName
ab, einschließlich Definition, Größe und Exporterstatus des Streams.
- Python
-
client = StreamManagerClient()
try:
stream_description = client.describe_message_stream(stream_name="StreamName")
if stream_description.export_statuses[0].error_message:
# The last export of export destination 0 failed with some error
# Here is the last sequence number that was successfully exported
stream_description.export_statuses[0].last_exported_sequence_number
if (stream_description.storage_status.newest_sequence_number >
stream_description.export_statuses[0].last_exported_sequence_number):
pass
# The end of the stream is ahead of the last exported sequence number
except StreamManagerException:
pass
# Properly handle errors.
except ConnectionError or asyncio.TimeoutError:
pass
# Properly handle errors.
Python-SDK-Referenz: describe_message_stream
- Java
-
try (final StreamManagerClient client = StreamManagerClientFactory.standard().build()) {
MessageStreamInfo description = client.describeMessageStream("StreamName");
String lastErrorMessage = description.getExportStatuses().get(0).getErrorMessage();
if (lastErrorMessage != null && !lastErrorMessage.equals("")) {
// The last export of export destination 0 failed with some error.
// Here is the last sequence number that was successfully exported.
description.getExportStatuses().get(0).getLastExportedSequenceNumber();
}
if (description.getStorageStatus().getNewestSequenceNumber() >
description.getExportStatuses().get(0).getLastExportedSequenceNumber()) {
// The end of the stream is ahead of the last exported sequence number.
}
} catch (StreamManagerException e) {
// Properly handle exception.
}
Java-SDK-Referenz: describeMessageStream
- Node.js
-
const client = new StreamManagerClient();
client.onConnected(async () => {
try {
const description = await client.describeMessageStream("StreamName");
const lastErrorMessage = description.exportStatuses[0].errorMessage;
if (lastErrorMessage) {
// The last export of export destination 0 failed with some error.
// Here is the last sequence number that was successfully exported.
description.exportStatuses[0].lastExportedSequenceNumber;
}
if (description.storageStatus.newestSequenceNumber >
description.exportStatuses[0].lastExportedSequenceNumber) {
// The end of the stream is ahead of the last exported sequence number.
}
} catch (e) {
// Properly handle errors.
}
});
client.onError((err) => {
// Properly handle connection errors.
// This is called only when the connection to the StreamManager server fails.
});
Node.js SDK-Referenz: describeMessageStream
Nachrichtenstream aktualisieren
Aktualisieren Sie die Eigenschaften eines vorhandenen Streams. Möglicherweise möchten Sie einen Stream aktualisieren, wenn sich Ihre Anforderungen ändern, nachdem der Stream erstellt wurde. Beispielsweise:
-
Fügen Sie eine neue Exportkonfiguration für ein AWS Cloud Ziel hinzu.
-
Erhöhen Sie die maximale Größe eines Streams, um zu ändern, wie Daten exportiert oder aufbewahrt werden. Beispielsweise kann die Stream-Größe in Kombination mit Ihrer Strategie für vollständige Einstellungen dazu führen, dass Daten gelöscht oder abgelehnt werden, bevor der Stream-Manager sie verarbeiten kann.
-
Pausieren und Fortsetzen von Exporten, z. B. wenn Exportaufgaben lange laufen und Sie Ihre Upload-Daten rationieren möchten.
Ihre Greengrass-Komponenten folgen diesem allgemeinen Prozess, um einen Stream zu aktualisieren:
-
Rufen Sie die Beschreibung des Streams ab.
-
Aktualisieren Sie die Zieleigenschaften für die entsprechenden MessageStreamDefinition
und untergeordneten Objekte.
-
Übergeben Sie das aktualisierte MessageStreamDefinition
. Stellen Sie sicher, dass Sie die vollständigen Objektdefinitionen für den aktualisierten Stream angeben. Undefinierte Eigenschaften setzen auf die Standardwerte zurück.
Sie können die Sequenznummer der Nachricht angeben, die als Startnachricht im Export verwendet werden soll.
Voraussetzungen
Für diesen Vorgang gelten die folgenden Anforderungen:
Beispiele
Der folgende Codeausschnitt aktualisiert den Stream mit dem Namen StreamName
. Es aktualisiert mehrere Eigenschaften eines Streams, der nach Kinesis Data Streams exportiert wird.
- Python
-
client = StreamManagerClient()
try:
message_stream_info = client.describe_message_stream(STREAM_NAME)
message_stream_info.definition.max_size=536870912
message_stream_info.definition.stream_segment_size=33554432
message_stream_info.definition.time_to_live_millis=3600000
message_stream_info.definition.strategy_on_full=StrategyOnFull.RejectNewData
message_stream_info.definition.persistence=Persistence.Memory
message_stream_info.definition.flush_on_write=False
message_stream_info.definition.export_definition.kinesis=
[KinesisConfig(
# Updating Export definition to add a Kinesis Stream configuration.
identifier=str(uuid.uuid4()), kinesis_stream_name=str(uuid.uuid4()))]
client.update_message_stream(message_stream_info.definition)
except StreamManagerException:
pass
# Properly handle errors.
except ConnectionError or asyncio.TimeoutError:
pass
# Properly handle errors.
Python-SDK-Referenz: updateMessageStream | MessageStreamDefinition
- Java
-
try (final StreamManagerClient client = GreengrassClientBuilder.streamManagerClient().build()) {
MessageStreamInfo messageStreamInfo = client.describeMessageStream(STREAM_NAME);
// Update the message stream with new values.
client.updateMessageStream(
messageStreamInfo.getDefinition()
.withStrategyOnFull(StrategyOnFull.RejectNewData) // Required. Updating Strategy on full to reject new data.
// Max Size update should be greater than initial Max Size defined in Create Message Stream request
.withMaxSize(536870912L) // Update Max Size to 512 MB.
.withStreamSegmentSize(33554432L) // Update Segment Size to 32 MB.
.withFlushOnWrite(true) // Update flush on write to true.
.withPersistence(Persistence.Memory) // Update the persistence to Memory.
.withTimeToLiveMillis(3600000L) // Update TTL to 1 hour.
.withExportDefinition(
// Optional. Choose where/how the stream is exported to the AWS Cloud.
messageStreamInfo.getDefinition().getExportDefinition().
// Updating Export definition to add a Kinesis Stream configuration.
.withKinesis(new ArrayList<KinesisConfig>() {{
add(new KinesisConfig()
.withIdentifier(EXPORT_IDENTIFIER)
.withKinesisStreamName("test"));
}})
);
} catch (StreamManagerException e) {
// Properly handle exception.
}
Java-SDK-Referenz: update_message_stream | MessageStreamDefinition
- Node.js
-
const client = new StreamManagerClient();
client.onConnected(async () => {
try {
const messageStreamInfo = await c.describeMessageStream(STREAM_NAME);
await client.updateMessageStream(
messageStreamInfo.definition
// Max Size update should be greater than initial Max Size defined in Create Message Stream request
.withMaxSize(536870912) // Default is 256 MB. Updating Max Size to 512 MB.
.withStreamSegmentSize(33554432) // Default is 16 MB. Updating Segment Size to 32 MB.
.withTimeToLiveMillis(3600000) // By default, no TTL is enabled. Update TTL to 1 hour.
.withStrategyOnFull(StrategyOnFull.RejectNewData) // Required. Updating Strategy on full to reject new data.
.withPersistence(Persistence.Memory) // Default is File. Update the persistence to Memory
.withFlushOnWrite(true) // Default is false. Updating to true.
.withExportDefinition(
// Optional. Choose where/how the stream is exported to the AWS Cloud.
messageStreamInfo.definition.exportDefinition
// Updating Export definition to add a Kinesis Stream configuration.
.withKinesis([new KinesisConfig().withIdentifier(uuidv4()).withKinesisStreamName(uuidv4())])
)
);
} catch (e) {
// Properly handle errors.
}
});
client.onError((err) => {
// Properly handle connection errors.
// This is called only when the connection to the StreamManager server fails.
});
Node.js SDK-Referenz: updateMessageStream | MessageStreamDefinition
Einschränkungen für die Aktualisierung von Streams
Beim Aktualisieren von Streams gelten die folgenden Einschränkungen. Sofern in der folgenden Liste nicht anders angegeben, werden Aktualisierungen sofort wirksam.
-
Sie können die Persistenz eines Streams nicht aktualisieren. Um dieses Verhalten zu ändern, löschen Sie den Stream und erstellen Sie einen Stream, der die neue Persistenzrichtlinie definiert.
-
Sie können die maximale Größe eines Streams nur unter den folgenden Bedingungen aktualisieren:
-
Die maximale Größe muss größer oder gleich der aktuellen Größe des Streams sein. Um diese Informationen zu finden, beschreiben Sie den Stream und überprüfen Sie dann den Speicherstatus des zurückgegebenen MessageStreamInfo
Objekts.
-
Die maximale Größe muss größer oder gleich der Segmentgröße des Streams sein.
-
Sie können die Stream-Segmentgröße auf einen Wert aktualisieren, der kleiner als die maximale Größe des Streams ist. Die aktualisierte Einstellung gilt für neue Segmente.
-
Aktualisierungen der Time to Live (TTL)-Eigenschaft gelten für neue Append-Operationen. Wenn Sie diesen Wert verringern, löscht Stream Manager möglicherweise auch vorhandene Segmente, die die TTL überschreiten.
-
Aktualisierungen der Strategie für die vollständige Eigenschaft gelten für neue Append-Operationen. Wenn Sie die Strategie zum Überschreiben der ältesten Daten festlegen, überschreibt Stream Manager möglicherweise auch vorhandene Segmente basierend auf der neuen Einstellung.
-
Aktualisierungen der Eigenschaft flush on write gelten für neue Nachrichten.
-
Aktualisierungen der Exportkonfigurationen gelten für neue Exporte. Die Aktualisierungsanforderung muss alle Exportkonfigurationen enthalten, die Sie unterstützen möchten. Andernfalls löscht der Stream-Manager sie.
-
Wenn Sie eine Exportkonfiguration aktualisieren, geben Sie die Kennung der Zielexportkonfiguration an.
-
Um eine Exportkonfiguration hinzuzufügen, geben Sie eine eindeutige Kennung für die neue Exportkonfiguration an.
-
Um eine Exportkonfiguration zu löschen, lassen Sie die Exportkonfiguration weg.
-
Um die Startsequenznummer einer Exportkonfiguration in einem Stream zu aktualisieren, müssen Sie einen Wert angeben, der kleiner als die neueste Sequenznummer ist. Um diese Informationen zu finden, beschreiben Sie den Stream und überprüfen Sie dann den Speicherstatus des zurückgegebenen MessageStreamInfo
Objekts.
Löschen eines Nachrichten-Streams
Löscht einen Stream. Wenn Sie einen Stream löschen, werden alle gespeicherten Daten für den Stream von der Festplatte gelöscht.
Voraussetzungen
Für diesen Vorgang gelten die folgenden Anforderungen:
Beispiele
Das folgende Snippet löscht den Stream mit dem Namen StreamName
.
- Python
-
client = StreamManagerClient()
try:
client.delete_message_stream(stream_name="StreamName")
except StreamManagerException:
pass
# Properly handle errors.
except ConnectionError or asyncio.TimeoutError:
pass
# Properly handle errors.
Python-SDK-Referenz: deleteMessageStream
- Java
-
try (final StreamManagerClient client = StreamManagerClientFactory.standard().build()) {
client.deleteMessageStream("StreamName");
} catch (StreamManagerException e) {
// Properly handle exception.
}
Java SDK-Referenz: delete_message_stream
- Node.js
-
const client = new StreamManagerClient();
client.onConnected(async () => {
try {
await client.deleteMessageStream("StreamName");
} catch (e) {
// Properly handle errors.
}
});
client.onError((err) => {
// Properly handle connection errors.
// This is called only when the connection to the StreamManager server fails.
});
Referenz zum Node.js SDK: deleteMessageStream
Weitere Informationen finden Sie auch unter