Gunakan StreamManagerClient untuk bekerja dengan aliran - AWS IoT Greengrass

Terjemahan disediakan oleh mesin penerjemah. Jika konten terjemahan yang diberikan bertentangan dengan versi bahasa Inggris aslinya, utamakan versi bahasa Inggris.

Gunakan StreamManagerClient untuk bekerja dengan aliran

Komponen Greengrass yang ditentukan oleh pengguna yang berjalan pada perangkat inti Greengrass dapat menggunakan objek StreamManagerClient di SDK Stream Manager untuk membuat aliran di stream manager dan kemudian berinteraksi dengan aliran tersebut. Ketika komponen menciptakan pengaliran, komponen tersebut menentukan tujuan AWS Cloud, prioritas, serta kebijakan ekspor dan retensi data lainnya untuk pengaliran tersebut. Untuk mengirim data ke stream manager, komponen menambahkan data ke aliran. Jika tujuan ekspor telah ditentukan untuk aliran tersebut, stream manager akan mengekspor aliran tersebut secara otomatis.

catatan

Biasanya, klien stream manager adalah komponen Greengrass yang ditetapkan pengguna. Jika kasus bisnis Anda memerlukannya, Anda juga dapat mengizinkan proses non-komponen yang berjalan pada inti Greengrass (misalnya, kontainer Docker) untuk berinteraksi dengan stream manager. Untuk informasi selengkapnya, lihat Autentikasi Klien.

Cuplikan dalam topik ini menunjukkan cara klien memanggil metode StreamManagerClient untuk bekerja dengan aliran. Untuk detail implementasi tentang metode dan argumennya, gunakan tautan ke referensi SDK terdaftar setelah setiap potongan.

Jika Anda menggunakan stream manager dalam fungsi Lambda, fungsi Lambda Anda harus memberi contoh StreamManagerClient di luar fungsi handler. Jika dipakai dalam handler, fungsi tersebut akan membuat client dan koneksi ke manajer pengaliran setiap kali dipanggil.

catatan

Jika Anda membuat contoh StreamManagerClient dalam handler, Anda harus secara tegas memanggil metode close() ketika client menyelesaikan pekerjaannya. Jika tidak, client akan membuat sambungan terbuka dan utas lain yang berjalan sampai skrip keluar.

StreamManagerClient mendukung operasi berikut:

Buat aliran pesan

Untuk membuat pengaliran, komponen Greengrass yang ditetapkan pengguna akan memanggil metode buat dan melewati objek MessageStreamDefinition. Objek ini menentukan nama unik untuk aliran tersebut dan menentukan bagaimana stream manager harus menangani data baru ketika ukuran aliran maksimum tercapai. Anda dapat menggunakan MessageStreamDefinition dan jenis datanya (seperti ExportDefinition, StrategyOnFull, dan Persistence) untuk menentukan properti pengaliran lainnya. Ini termasuk:

  • Target AWS IoT Analytics, Kinesis Data Streams, AWS IoT SiteWise, dan tujuan Amazon S3 untuk ekspor otomatis. Untuk informasi selengkapnya, lihat Ekspor konfigurasi untuk tujuan AWS Cloud yang didukung.

  • Prioritas ekspor. Stream manager mengekspor aliran prioritas yang lebih tinggi sebelum aliran prioritas lebih rendah.

  • Ukuran batch maksimum dan interval batch untuk AWS IoT Analytics, Kinesis Data Streams, dan tujuan AWS IoT SiteWise. Stream manager mengekspor pesan ketika salah satu kondisi terpenuhi.

  • Time-to-live (TTL). Jumlah waktu untuk menjamin bahwa data aliran tersedia untuk diproses. Anda harus memastikan bahwa data dapat dikonsumsi dalam periode waktu ini. Ini bukan kebijakan penghapusan. Data mungkin tidak segera dihapus setelah periode TTL.

  • Ketekunan aliran. Pilih untuk menyimpan aliran ke sistem file untuk mempertahankan data di seluruh restart inti atau menyimpan aliran dalam memori.

  • Memulai nomor urut. Tentukan nomor urutan pesan yang akan digunakan sebagai pesan awal dalam ekspor.

Untuk informasi lebih lanjut tentang MessageStreamDefinition, lihat referensi SDK untuk bahasa target Anda:

catatan

StreamManagerClient juga menyediakan target tujuan yang dapat Anda gunakan untuk mengekspor aliran ke server HTTP. Target ini ditujukan untuk tujuan pengujian saja. Target ini tidak stabil atau didukung untuk digunakan di lingkungan produksi.

Setelah stream dibuat, komponen Greengrass Anda dapat menambahkan pesan ke aliran tersebut untuk mengirim data untuk ekspor dan membaca pesan dari aliran tersebut untuk pemrosesan lokal. Jumlah aliran yang Anda buat tergantung pada kemampuan perangkat keras dan kasus bisnis Anda. Salah satu strateginya adalah membuat aliran untuk setiap saluran target di AWS IoT Analytics atau aliran data Kinesis, meskipun Anda dapat menentukan beberapa target untuk suatu aliran. Aliran memiliki masa pakai yang tahan lama.

Persyaratan

Komponen ini memiliki persyaratan sebagai berikut:

  • versi SDK Stream am Manager. Python: Java: 1.1.0 | Node.js |: 1.1.0

Contoh

Potongan berikut menciptakan aliran bernama StreamName. Potongan ini menentukan sifat aliran di MessageStreamDefinition dan jenis data bawahan.

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.

Referensi Python SDK: 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. }

Referensi Java SDK: 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. });

Referensi Node.js SDK: createMessageStream| MessageStreamDefinition

Untuk informasi lebih lanjut tentang konfigurasi tujuan ekspor, lihatEkspor konfigurasi untuk tujuan AWS Cloud yang didukung.

Tambahkan pesan

Untuk mengirim data ke stream manager untuk ekspor, komponen Greengrass Anda akan menambahkan data ke target aliran. Tujuan ekspor menentukan jenis data yang akan lolos ke metode ini.

Persyaratan

Komponen ini memiliki persyaratan sebagai berikut:

  • versi SDK Stream am Manager. Python: Java: 1.1.0 | Node.js |: 1.1.0

Contoh

AWS IoT Analytics atau tujuan ekspor Kinesis Data Streams

Potongan berikut menambahkan pesan ke aliran bernama StreamName. Untuk AWS IoT Analytics atau tujuan Kinesis Data Streams, komponen Greengrass Anda menambahkan sekumpulan data.

Potongan ini memiliki persyaratan sebagai berikut:

  • versi SDK Stream am Manager. Python: Java: 1.1.0 | Node.js |: 1.1.0

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.

Referensi SDK Python: 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. }

Referensi SDK Java: 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. });

Referensi SDK Node.js: appendMessage

Tujuan ekspor AWS IoT SiteWise

Potongan berikut menambahkan pesan ke aliran bernama StreamName. Untuk tujuan AWS IoT SiteWise, komponen Greengrass Anda akan menambahkan objek PutAssetPropertyValueEntry berserial. Untuk informasi selengkapnya, lihat Ekspor ke AWS IoT SiteWise.

catatan

Saat Anda mengirim data ke AWS IoT SiteWise, data Anda harus memenuhi persyaratan tindakan BatchPutAssetPropertyValue. Untuk informasi selengkapnya, lihat BatchPutAssetPropertyValue dalam Referensi API AWS IoT SiteWise.

Potongan ini memiliki persyaratan sebagai berikut:

  • versi SDK Stream am Manager. Python: Java: 1.1.0 | Node.js |: 1.1.0

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.

Referensi Python SDK: 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. }

Referensi Java SDK: 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. });

Referensi Node.js SDK: appendMessage | PutAssetPropertyValueEntry

Tujuan ekspor Amazon S3

Potongan berikut menambahkan tugas ekspor ke aliran bernama StreamName. Untuk tujuan Amazon S3, komponen Greengrass Anda menambahkan serial objek S3ExportTaskDefinition yang berisi informasi tentang file input sumber dan target objek Amazon S3. Jika objek tertentu tidak ada, Stream Manager akan membuatnya untuk Anda. Untuk informasi selengkapnya, lihat Mengekspor ke Amazon S3.

Cuplikan ini memiliki persyaratan sebagai berikut:

  • versi SDK Stream am Manager. Python: Java: 1.1.0 | Node.js |: 1.1.0

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.

Referensi Python SDK: 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. }

Referensi Java SDK: 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. });

Referensi SDK Node.js: appendMessage | S3ExportTaskDefinition

Baca pesan

Baca pesan dari suatu aliran.

Persyaratan

Komponen ini memiliki persyaratan sebagai berikut:

  • versi SDK Stream am Manager. Python: Java: 1.1.0 | Node.js |: 1.1.0

Contoh

Potongan berikut membaca pesan dari aliran bernama StreamName. Metode membaca mengambil objek ReadMessagesOptions opsional yang menentukan urutan nomor untuk memulai membaca, jumlah minimum dan maksimum yang dibaca, dan batas waktu untuk membaca pesan.

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.

Referensi Python SDK: 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. }

Referensi Java SDK: 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. });

Referensi Node.js SDK: readMessages | ReadMessagesOptions

Daftar aliran

Dapatkan daftar aliran di stream manager.

Persyaratan

Komponen ini memiliki persyaratan sebagai berikut:

  • versi SDK Stream am Manager. Python: Java: 1.1.0 | Node.js |: 1.1.0

Contoh

Potongan berikut mendapat daftar aliran (dengan nama) di stream manager.

Python
client = StreamManagerClient() try: stream_names = client.list_streams() except StreamManagerException: pass # Properly handle errors. except ConnectionError or asyncio.TimeoutError: pass # Properly handle errors.

Referensi SDK Python: list_streams

Java
try (final StreamManagerClient client = StreamManagerClientFactory.standard().build()) { List<String> streamNames = client.listStreams(); } catch (StreamManagerException e) { // Properly handle exception. }

Referensi SDK Java: 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. });

Referensi SDK Node.js: listStreams

Jelaskan aliran pesan

Dapatkan metadata tentang aliran, termasuk definisi, ukuran, dan status ekspor aliran.

Persyaratan

Komponen ini memiliki persyaratan sebagai berikut:

  • versi SDK Stream am Manager. Python: Java: 1.1.0 | Node.js |: 1.1.0

Contoh

Potongan berikut mendapat metadata tentang aliran bernama StreamName, termasuk status definisi, ukuran, dan pengekspor aliran.

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.

Referensi SDK Python: 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. }

Referensi SDK Java: 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. });

Referensi Node.js SDK: describeMessageStream

Perbarui aliran pesan

Perbarui properti dari aliran yang ada. Anda mungkin ingin memperbarui aliran jika kebutuhan Anda berubah setelah aliran dibuat. Misalnya:

  • Tambahkan konfigurasi ekspor baru untuk tujuan AWS Cloud.

  • Tingkatkan ukuran maksimum aliran untuk mengubah cara data diekspor atau disimpan. Misalnya, ukuran aliran yang dikombinasikan dengan strategi Anda pada pengaturan penuh dapat mengakibatkan data dihapus atau ditolak sebelum stream manager dapat memprosesnya.

  • Jeda dan lanjutkan ekspor; misalnya, jika tugas ekspor berjalan lama dan Anda ingin membagi data unggahan Anda.

Komponen Greengrass Anda mengikuti proses tingkat tinggi ini untuk memperbarui aliran:

  1. Dapatkan deskripsi aliran.

  2. Perbarui properti target pada MessageStreamDefinition dan objek bawahan yang sesuai.

  3. Lewati MessageStreamDefinition yang diperbarui. Pastikan untuk menyertakan definisi objek lengkap untuk aliran yang diperbarui. Properti yang tidak terdefinisi kembali ke nilai default.

    Anda dapat menentukan nomor urutan pesan yang akan digunakan sebagai pesan awal dalam ekspor.

Persyaratan

Komponen ini memiliki persyaratan sebagai berikut:

  • versi SDK Stream am Manager. Python: Java: 1.1.0 | Node.js |: 1.1.0

Contoh

Potongan berikut menciptakan aliran bernama StreamName. Potongan ini memperbarui beberapa properti aliran yang mengekspor ke Kinesis Data Streams.

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.

Referensi Python SDK: 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. }

Referensi SDK Java: 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. });

Referensi Node.js SDK: updateMessageStream| MessageStreamDefinition

Kendala untuk memperbarui aliran

Kendala berikut berlaku saat memperbarui aliran. Kecuali tercantum dalam daftar berikut, pembaruan berlaku segera.

  • Anda tidak dapat memperbarui kegigihan aliran. Untuk mengubah perilaku ini, hapus aliran dan buat Stream yang menentukan kebijakan kegigihan baru.

  • Anda dapat memperbarui ukuran maksimum aliran hanya dalam kondisi berikut:

    • Ukuran maksimum harus lebih besar atau sama dengan ukuran aliran. Untuk menemukan informasi ini, jelaskan aliran dan kemudian periksa status penyimpananMessageStreamInfo objek yang dikembalikan.

    • Ukuran maksimum harus lebih besar atau sama dengan ukuran segmen aliran.

  • Anda dapat memperbarui ukuran segmen aliran ke nilai kurang dari ukuran maksimum aliran tersebut. Pengaturan yang diperbarui berlaku untuk segmen baru.

  • Pembaruan ke properti waktu untuk tayang (TTL) berlaku untuk operasi append yang baru. Jika Anda mengurangi nilai ini, stream manager juga dapat menghapus segmen yang ada yang melebihi TTL.

  • Pembaruan untuk strategi tersebut pada properti penuh berlaku untuk operasi append yang baru. Jika Anda menetapkan strategi untuk menimpa data tertua, stream manager juga dapat menimpa segmen yang ada berdasarkan pengaturan yang baru.

  • Pembaruan untuk properti flush on write berlaku untuk pesan baru.

  • Pembaruan untuk konfigurasi ekspor berlaku untuk ekspor baru. Permintaan pembaruan harus mencakup semua konfigurasi ekspor yang ingin Anda dukung. Jika tidak, stream manager akan menghapusnya.

    • Saat Anda memperbarui konfigurasi ekspor, tentukan pengenal konfigurasi ekspor target.

    • Untuk menambahkan konfigurasi ekspor, tentukan pengenal unik untuk konfigurasi ekspor baru.

    • Untuk menghapus konfigurasi ekspor, hapus konfigurasi ekspor.

  • Untuk memperbarui nomor urutan awal konfigurasi ekspor di suatu aliran, Anda harus menentukan nilai yang kurang dari nomor urut terbaru. Untuk menemukan informasi ini, jelaskan aliran dan kemudian periksa status penyimpananMessageStreamInfo objek yang dikembalikan.

Hapus aliran pesan

Hapus aliran. Bila Anda menghapus suatu aliran, semua data yang tersimpan untuk aliran itu akan dihapus dari disk.

Persyaratan

Komponen ini memiliki persyaratan sebagai berikut:

  • versi SDK Stream am Manager. Python: Java: 1.1.0 | Node.js |: 1.1.0

Contoh

Potongan berikut menghapus aliran bernama 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.

Referensi Python SDK: deleteMessageStream

Java
try (final StreamManagerClient client = StreamManagerClientFactory.standard().build()) { client.deleteMessageStream("StreamName"); } catch (StreamManagerException e) { // Properly handle exception. }

Referensi SDK Java: 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. });

Referensi Node.js SDK: deleteMessageStream

Lihat juga