Terjemahan disediakan oleh mesin penerjemah. Jika konten terjemahan yang diberikan bertentangan dengan versi bahasa Inggris aslinya, utamakan versi bahasa Inggris. # Kelola aliran data di perangkat inti Greengrass AWS IoT Greengrass stream manager membuatnya lebih efisien dan andal untuk mentransfer data IoT volume tinggi ke file. AWS Cloud Manajer aliran memproses aliran data pada AWS IoT Greengrass Core sebelum mengekspornya ke file. AWS Cloud Stream manager terintegrasi dengan skenario tepi umum, seperti inferensi pembelajaran mesin (ML), di mana perangkat AWS IoT Greengrass Core memproses dan menganalisis data sebelum mengekspor data ke AWS Cloud atau tujuan penyimpanan lokal. Stream manager menyediakan antarmuka umum untuk menyederhanakan pengembangan komponen kustom sehingga Anda tidak perlu membangun fungsi manajemen aliran kustom. Komponen Anda dapat menggunakan mekanisme standar untuk memproses aliran volume tinggi dan mengelola kebijakan penyimpanan data lokal. Anda dapat menentukan kebijakan untuk jenis penyimpanan, ukuran, dan retensi data untuk setiap aliran guna mengontrol cara stream manager memproses dan mengekspor data. Stream manager bekerja di lingkungan dengan konektivitas yang terputus-putus atau terbatas. Anda dapat menentukan penggunaan bandwidth, perilaku batas waktu, dan bagaimana AWS IoT Greengrass Core menangani data aliran saat terhubung atau terputus. Anda juga dapat menetapkan prioritas untuk mengontrol urutan di mana inti AWS IoT Greengrass mengekspor aliran ke AWS Cloud. Hal ini memungkinkan Anda untuk menangani data penting lebih cepat dari data lainnya. Anda dapat mengonfigurasi manajer aliran untuk secara otomatis mengekspor data ke AWS Cloud penyimpanan atau pemrosesan dan analisis lebih lanjut. Stream manager mendukung ekspor ke tujuan AWS Cloud berikut: + Saluran di AWS IoT Analytics. AWS IoT Analytics memungkinkan Anda melakukan analisis lanjutan pada data Anda untuk membantu membuat keputusan bisnis dan meningkatkan model pembelajaran mesin. Untuk informasi selengkapnya, lihat [Apa itu AWS IoT Analytics?](https://docs.aws.amazon.com/iotanalytics/latest/userguide/welcome.html) dalam *Panduan Pengguna AWS IoT Analytics *. + Pengaliran di Amazon Kinesis Data Streams Anda dapat menggunakan Kinesis Data Streams untuk mengumpulkan data volume tinggi dan memuatnya ke gudang data atau cluster. MapReduce Untuk informasi selengkapnya, lihat [Apa yang dimaksud dengan Amazon Kinesis Data Streams?](https://docs.aws.amazon.com/streams/latest/dev/what-is-this-service.html) dalam *Panduan Developer Amazon Kinesis Data Streams*. + Properti aset di AWS IoT SiteWise. AWS IoT SiteWise memungkinkan Anda mengumpulkan, mengatur, dan menganalisis data dari peralatan industri dalam skala besar. Untuk informasi lebih lanjut, lihat [Apa itu AWS IoT SiteWise?](https://docs.aws.amazon.com/iot-sitewise/latest/userguide/what-is-sitewise.html) dalam *AWS IoT SiteWise User Guide*. + Objek di Amazon Simple Storage Service Amazon S3. Sebagai contoh, Anda dapat menggunakan Amazon S3 untuk menyimpan dan mengambil sejumlah besar data. Untuk informasi selengkapnya, lihat [Apa itu Amazon S3](https://docs.aws.amazon.com/AmazonS3/latest/dev/Welcome.html) dalam *Panduan Developer Amazon Simple Storage Service*. ## Alur kerja manajemen aliran Aplikasi IoT Anda berinteraksi dengan stream manager melalui SDK Stream Manager. Dalam alur kerja sederhana, komponen pada AWS IoT Greengrass inti mengkonsumsi data IoT, seperti suhu deret waktu dan metrik tekanan. Komponen tersebut mungkin memfilter atau mengompresi data, dan kemudian memanggil SDK Manajer Pengaliran untuk menuliskan data ke pengaliran di manajer pengaliran. Manajer aliran dapat mengekspor aliran ke aliran AWS Cloud secara otomatis berdasarkan kebijakan yang Anda tetapkan untuk aliran. Komponen juga dapat mengirim data langsung ke basis data lokal atau repositori penyimpanan. Aplikasi IoT Anda dapat mencakup beberapa komponen kustom yang membaca atau menulis ke pengaliran. Komponen-komponen ini dapat membaca dan menulis ke aliran untuk memfilter, mengumpulkan, dan menganalisis data pada perangkat AWS IoT Greengrass inti. Hal ini memungkinkan untuk merespons dengan cepat acara lokal dan mengekstrak informasi berharga sebelum transfer data dari inti ke tujuan lokal AWS Cloud atau lokal. Untuk memulai, terapkan komponen pengelola aliran ke perangkat AWS IoT Greengrass inti Anda. Dalam deployment, konfigurasi parameter komponen stream manager untuk menentukan pengaturan yang berlaku untuk semua aliran pada perangkat inti Greengrass. Gunakan parameter ini untuk mengontrol cara stream manager menyimpanmem, proses, dan mengekspor aliran berdasarkan kebutuhan bisnis dan kendala lingkungan Anda. Setelah Anda mengonfigurasi manajer pengaliran, Anda dapat membuat dan men-deploy aplikasi IoT Anda. Hal ini biasanya merupakan komponen kustom yang menggunakan `StreamManagerClient` di SDK Stream Manager untuk membuat dan berinteraksi dengan aliran. Saat membuat aliran, Anda dapat menentukan kebijakan per aliran, seperti tujuan ekspor, prioritas, dan kegigihan. ## Persyaratan Persyaratan berikut berlaku untuk penggunaan stream manager: + Stream manager membutuhkan minimal 70 MB RAM selain perangkat lunak AWS IoT Greengrass Core. Kebutuhan memori total Anda tergantung pada beban kerja Anda. + AWS IoT Greengrass komponen harus menggunakan Stream Manager SDK untuk berinteraksi dengan manajer aliran. SDK Stream Manager tersedia dalam bahasa berikut: + [SDK for Java Stream Manager](https://github.com/aws-greengrass/aws-greengrass-stream-manager-sdk-java/) (v1.1.0 atau yang lebih baru) + [SDK for Node.js Stream Manager](https://github.com/aws-greengrass/aws-greengrass-stream-manager-sdk-js/) (v1.1.0 atau yang lebih baru) + [SDK for Python Stream Manager](https://github.com/aws-greengrass/aws-greengrass-stream-manager-sdk-python/) (v1.1.0 atau yang lebih baru) + AWS IoT Greengrass components harus menentukan komponen stream manager (`aws.greengrass.StreamManager`) sebagai dependensi dalam resepnya untuk menggunakan stream manager. **catatan** Jika Anda menggunakan pengelola aliran untuk mengekspor data ke cloud, Anda tidak dapat memutakhirkan versi 2.0.7 komponen pengelola aliran ke versi antara v2.0.8 dan v2.0.11. Jika Anda menerapkan pengelola aliran untuk pertama kalinya, kami sangat menyarankan agar Anda menerapkan versi terbaru komponen pengelola aliran. + Jika menentukan tujuan AWS Cloud ekspor untuk streaming, Anda harus membuat target ekspor dan memberikan izin akses dalam peran perangkat [Greengrass](device-service-role.md). Tergantung pada tujuan, persyaratan lain mungkin juga berlaku. Untuk informasi lebih lanjut, lihat: + [AWS IoT Analytics saluran](stream-export-configurations.md#export-to-iot-analytics) + [Amazon Kinesis data streams](stream-export-configurations.md#export-to-kinesis) + [AWS IoT SiteWise properti aset](stream-export-configurations.md#export-to-iot-sitewise) + [Objek Amazon S3](stream-export-configurations.md#export-to-s3) Anda bertanggung jawab untuk memelihara AWS Cloud sumber daya ini. ## Keamanan data Bila Anda menggunakan stream manager, perhatikan pertimbangan keamanan berikut ini. ### Keamanan data lokal AWS IoT Greengrass tidak mengenkripsi data aliran saat istirahat atau dalam transit antara komponen lokal pada perangkat inti. + **Data at rest**. Data pengaliran disimpan secara lokal dalam suatu direktori penyimpanan. Untuk keamanan data, AWS IoT Greengrass bergantung pada izin file dan enkripsi full-disk, jika diaktifkan. Anda dapat menggunakan parameter opsional [STREAM\$1MANAGER\$1STORE\$1ROOT\$1DIR](configure-stream-manager.md#STREAM_MANAGER_STORE_ROOT_DIR) untuk menentukan direktori penyimpanan. Jika Anda mengubah parameter ini nanti untuk menggunakan direktori penyimpanan yang berbeda, AWS IoT Greengrass tidak menghapus direktori penyimpanan sebelumnya atau isinya. + **Data dalam transit secara lokal**. AWS IoT Greengrass tidak mengenkripsi data aliran dalam transit lokal antara sumber data, AWS IoT Greengrass komponen, Stream Manager SDK, dan manajer aliran. + **Data dalam perjalanan ke AWS Cloud**. Aliran data diekspor oleh manajer aliran untuk AWS Cloud menggunakan enkripsi klien AWS layanan standar dengan Transport Layer Security (TLS). ### Autentikasi Klien Klien stream manager menggunakan SDK Stream Manager untuk berkomunikasi dengan stream manager. Ketika autentikasi klien diaktifkan, hanya komponen Greengrass yang dapat berinteraksi dengan aliran di stream manager. Ketika autentikasi klien dinonaktifkan, setiap proses yang berjalan pada perangkat inti Greengrass dapat berinteraksi dengan aliran di stream manager. Anda harus menonaktifkan autentikasi hanya jika kasus bisnis Anda memerlukannya. Anda menggunakan parameter [STREAM\$1MANAGER\$1AUTHENTICATE\$1CLIENT](configure-stream-manager.md#STREAM_MANAGER_AUTHENTICATE_CLIENT) untuk mengatur mode autentikasi klien. Anda dapat mengonfigurasikan parameter ini saat Anda men-deploy komponen stream manager tersebut ke perangkat inti Anda. **** |   | Diaktifkan | Dinonaktifkan | | --- | --- | --- | | Nilai parameter | `true` (default dan disarankan) | `false` | | Klien yang diizinkan | Komponen Greengrass pada perangkat inti | Komponen Greengrass pada perangkat inti Proses lain yang berjalan di perangkat inti Greengrass | ## Lihat juga + [Konfigurasikan pengelola AWS IoT Greengrass aliran](configure-stream-manager.md) + [Gunakan StreamManagerClient untuk bekerja dengan aliran](work-with-streams.md) + [Ekspor konfigurasi untuk tujuan yang didukung AWS Cloud](stream-export-configurations.md) # Konfigurasikan pengelola AWS IoT Greengrass aliran Pada perangkat inti Greengrass, stream manager dapat menyimpan, memproses, dan mengekspor data perangkat IoT. Strram manager menyediakan parameter yang Anda gunakan untuk mengonfigurasi pengaturan waktu aktif. Pengaturan ini berlaku untuk semua aliran pada perangkat inti Greengrass. Anda dapat menggunakan AWS IoT Greengrass konsol atau API untuk mengonfigurasi pengaturan pengelola aliran saat menerapkan komponen. Perubahan berlaku setelah deployment selesai. ## Parameter stream manager Stream manager menyediakan parameter berikut yang dapat Anda konfigurasikan saat Anda men-deploy komponen tersebut ke perangkat inti Anda. Semua parameter bersifat opsional. **Direktori penyimpanan** Nama parameter: `STREAM_MANAGER_STORE_ROOT_DIR` Jalur absolut dari folder lokal yang digunakan untuk menyimpan aliran. Nilai ini harus dimulai dengan garis miring ke depan (misalnya, `/data`). Anda harus menentukan folder yang ada, dan [pengguna sistem yang menjalankan komponen manajer aliran](configure-greengrass-core-v2.md#configure-component-user) harus memiliki izin untuk membaca dan menulis ke folder ini. Misalnya, Anda dapat menjalankan perintah berikut untuk membuat dan mengkonfigurasi folder`/var/greengrass/streams`, yang Anda tentukan sebagai folder root stream manager. Perintah-perintah ini memungkinkan pengguna sistem default`ggc_user`,, untuk membaca dan menulis ke folder ini. ``` sudo mkdir /var/greengrass/streams sudo chown ggc_user /var/greengrass/streams sudo chmod 700 /var/greengrass/streams ``` Untuk informasi tentang cara mengamankan data aliran, lihat [Keamanan data lokal](manage-data-streams.md#stream-manager-security-stream-data). Default: `/greengrass/v2/work/aws.greengrass.StreamManager` **Port server** Nama parameter: `STREAM_MANAGER_SERVER_PORT` Nomor port lokal yang digunakan untuk berkomunikasi dengan stream manager. Default-nya adalah `8088`. Anda dapat menentukan `0` untuk menggunakan port yang tersedia acak. **Otentikasi klien** Nama parameter: `STREAM_MANAGER_AUTHENTICATE_CLIENT` Menunjukkan apakah klien harus diautentikasi untuk berinteraksi dengan stream manager. Semua interaksi antara klien dan stream manager dikendalikan oleh Stream Manager SDK. Parameter ini menentukan klien mana yang dapat memanggil Manajer Pengaliran untuk bekerja dengan pengaliran. Untuk informasi selengkapnya, lihat [Autentikasi Klien](manage-data-streams.md#stream-manager-security-client-authentication). Nilai yang valid adalah `true` atau `false`. Default-nya adalah `true` (direkomendasikan). + `true`. Memungkinkan hanya komponen Greengrass sebagai klien. Komponen menggunakan protokol AWS IoT Greengrass Inti internal untuk mengautentikasi dengan Stream Manager SDK. + `false`. Memungkinkan proses apa pun yang berjalan pada AWS IoT Greengrass Core menjadi klien. Jangan tetapkan nilai ke `false` kecuali kasus bisnis Anda memerlukannya. Misalnya, gunakan `false` hanya jika proses non-komponen pada perangkat inti harus berkomunikasi langsung dengan stream manager. **Bandwidth maksimum** Nama parameter: `STREAM_MANAGER_EXPORTER_MAX_BANDWIDTH` Bandwidth maksimum rata-rata (dalam kilobit per detik) yang dapat digunakan untuk mengekspor data. Default ini memungkinkan penggunaan bandwidth yang tersedia tanpa batas. **Ukuran kolam benang** Nama parameter: `STREAM_MANAGER_EXPORTER_THREAD_POOL_SIZE` Jumlah maksimum utas aktif yang dapat digunakan untuk mengekspor data. Default-nya adalah `5`. Ukuran optimal tergantung pada perangkat keras Anda, volume aliran, dan jumlah yang direncanakan dari aliran ekspor. Jika kecepatan ekspor lambat, Anda dapat menyesuaikan pengaturan ini untuk menemukan ukuran optimal untuk perangkat keras dan kasus bisnis Anda. CPU dan memori perangkat keras inti Anda merupakan faktor pembatas. Untuk memulai, Anda dapat mencoba menetapkan nilai ini sama dengan jumlah inti prosesor pada perangkat. Hati-hati untuk tidak menetapkan ukuran yang lebih tinggi dari yang dapat didukung perangkat keras Anda. Setiap aliran mengonsumsi sumber daya perangkat keras, jadi cobalah untuk membatasi jumlah aliran ekspor pada perangkat yang dibatasi. **Argumen JVM** Nama parameter: `JVM_ARGS` Argumen Mesin Virtual Java kustom yang akan disampaikan ke stream manager saat startup. Beberapa argumen harus dipisahkan oleh spasi. Gunakan parameter ini hanya ketika Anda harus menimpa pengaturan default yang digunakan oleh JVM. Misalnya, Anda mungkin perlu meningkatkan ukuran tumpukan default jika Anda berencana mengekspor sejumlah besar pengaliran. **Tingkat pencatatan** Nama parameter: `LOG_LEVEL` Level logging untuk komponen. Pilih dari tingkat log berikut, yang tercantum di sini dalam urutan tingkat: + `TRACE` + `DEBUG` + `INFO` + `WARN` + `ERROR` Default: `INFO` **Ukuran minimum untuk unggahan multipart** Nama parameter: `STREAM_MANAGER_EXPORTER_S3_DESTINATION_MULTIPART_UPLOAD_MIN_PART_SIZE_BYTES` Ukuran minimum (dalam byte) dari bagian dalam unggahan multipart ke Amazon S3. Stream manager menggunakan pengaturan ini dan ukuran file inputnya untuk menentukan bagaimana melakukan batch data dalam permintaan PUT multipart. Nilai minimum dan default adalah `5242880` byte (5 MB). Stream manager menggunakan properti `sizeThresholdForMultipartUploadBytes` aliran tersebut untuk menentukan apakah akan mengekspor ke Amazon S3 sebagai unggahan tunggal atau multipart. Komponen Greengrass yang ditentukan pengguna mengatur ambang batas ini ketika membuat aliran yang mengekspor ke Amazon S3. Ambang batas default adalah 5 MB. ## Lihat juga + [Kelola aliran data di perangkat inti Greengrass](manage-data-streams.md) + [Gunakan StreamManagerClient untuk bekerja dengan aliran](work-with-streams.md) + [Ekspor konfigurasi untuk tujuan yang didukung AWS Cloud](stream-export-configurations.md) # Buat komponen kustom yang menggunakan stream manager Gunakan manajer pengaliran pada komponen Greengrass kustom untuk menyimpan, memproses, dan mengekspor data perangkat IoT. Gunakan prosedur dan contoh di bagian ini untuk membuat resep komponen, artefak, dan aplikasi yang bekerja dengan stream manager. Untuk informasi lebih lanjut tentang cara mengembangkan dan menguji komponen, lihat [Buat AWS IoT Greengrass komponen](create-components.md). **Topics** + [Tentukan resep komponen yang menggunakan stream manager](#stream-manager-recipes) + [Hubungkan ke manajer pengaliran dalam kode aplikasi](#connect-to-stream-manager) ## Tentukan resep komponen yang menggunakan stream manager Untuk menggunakan stream manager dalam komponen kustom, Anda harus menentukan komponen `aws.greengrass.StreamManager` sebagai dependensi. Anda juga harus menyediakan SDK Stream Manager. Selesaikan tugas berikut untuk mengunduh dan menggunakan SDK Manajer Pengaliran dalam bahasa pilihan Anda. ### Gunakan SDK for Java Stream Manager SDK for Java Stream Manager tersedia sebagai file JAR yang dapat Anda gunakan untuk mengompilasi komponen Anda. Kemudian, Anda dapat membuat JAR aplikasi yang mencakup SDK Stream Manager, menentukan JAR aplikasi sebagai artefak komponen, dan menjalankan JAR aplikasi dalam siklus hidup komponen. **Untuk menggunakan SDK for Java Stream Manager** 1. Unduh [file JAR SDK for Java Stream Manager](https://github.com/aws-greengrass/aws-greengrass-stream-manager-sdk-java/blob/main/sdk/aws-greengrass-stream-manager-sdk-java.jar). 1. Lakukan salah satu dari hal berikut untuk membuat artefak komponen dari aplikasi Java Anda dan file JAR SDK Stream Manager: + Bangun komponen Anda sebagai file JAR yang mencakup JAR SDK Stream Manager, dan jalankan file JAR ini dalam resep komponen Anda. + Tentukan JAR SDK Stream Manager sebagai artefak komponen. Tambahkan artefak itu ke classpath ketika Anda menjalankan aplikasi Anda dalam resep komponen Anda. Resep komponen Anda mungkin terlihat serupa dengan contoh berikut ini. Komponen ini menjalankan versi modifikasi dari contoh [StreamManagerS3.java](https://github.com/aws-greengrass/aws-greengrass-stream-manager-sdk-java/blob/main/samples/StreamManagerS3/src/main/java/com/amazonaws/greengrass/examples/StreamManagerS3.java), yang `StreamManagerS3.jar` menyertakan JAR SDK Stream Manager. ------ #### [ JSON ] ``` { "RecipeFormatVersion": "2020-01-25", "ComponentName": "com.example.StreamManagerS3Java", "ComponentVersion": "1.0.0", "ComponentDescription": "Uses stream manager to upload a file to an S3 bucket.", "ComponentPublisher": "Amazon", "ComponentDependencies": { "aws.greengrass.StreamManager": { "VersionRequirement": "^2.0.0" } }, "Manifests": [ { "Lifecycle": { "Run": "java -jar {artifacts:path}/StreamManagerS3.jar" }, "Artifacts": [ { "URI": "s3://amzn-s3-demo-bucket/artifacts/com.example.StreamManagerS3Java/1.0.0/StreamManagerS3.jar" } ] } ] } ``` ------ #### [ YAML ] ``` --- RecipeFormatVersion: '2020-01-25' ComponentName: com.example.StreamManagerS3Java ComponentVersion: 1.0.0 ComponentDescription: Uses stream manager to upload a file to an S3 bucket. ComponentPublisher: Amazon ComponentDependencies: aws.greengrass.StreamManager: VersionRequirement: "^2.0.0" Manifests: - Lifecycle: Run: java -jar {artifacts:path}/StreamManagerS3.jar Artifacts: - URI: s3://amzn-s3-demo-bucket/artifacts/com.example.StreamManagerS3Java/1.0.0/StreamManagerS3.jar ``` ------ Untuk informasi lebih lanjut tentang cara mengembangkan dan menguji komponen, lihat [Buat AWS IoT Greengrass komponen](create-components.md). ### Gunakan SDK for Python Stream Manager SDK for Python Stream Manager tersedia sebagai kode sumber yang dapat Anda sertakan dalam komponen Anda. Buat file ZIP dari SDK Stream Manager, tentukan file ZIP sebagai artefak komponen, dan instal persyaratan SDK dalam siklus hidup komponen. **Untuk menggunakan SDK for Python Stream Manager** 1. Kloning atau unduh repositori [aws-greengrass-stream-manager-sdk-python](https://github.com/aws-greengrass/aws-greengrass-stream-manager-sdk-python). ``` git clone git@github.com:aws-greengrass/aws-greengrass-stream-manager-sdk-python.git ``` 1. Buat file ZIP yang berisi folder `stream_manager`, yang berisi kode sumber SDK for Python Manajer Pengaliran. Anda dapat menyediakan file ZIP ini sebagai artefak komponen yang dibuka ritsletingnya oleh perangkat lunak AWS IoT Greengrass Core saat menginstal komponen Anda. Lakukan hal-hal berikut: 1. Buka folder yang berisi repositori yang Anda kloning atau download pada langkah sebelumnya. ``` cd aws-greengrass-stream-manager-sdk-python ``` 1. Zip folder `stream_manager` ke dalam file ZIP bernama `stream_manager_sdk.zip`. ------ #### [ Linux or Unix ] ``` zip -rv stream_manager_sdk.zip stream_manager ``` ------ #### [ Windows Command Prompt (CMD) ] ``` tar -acvf stream_manager_sdk.zip stream_manager ``` ------ #### [ PowerShell ] ``` Compress-Archive stream_manager stream_manager_sdk.zip ``` ------ 1. Verifikasi bahwa file `stream_manager_sdk.zip` berisi folder `stream_manager` dan isinya. Jalankan perintah berikut untuk menampilkan konten file ZIP. ------ #### [ Linux or Unix ] ``` unzip -l stream_manager_sdk.zip ``` ------ #### [ Windows Command Prompt (CMD) ] ``` tar -tf stream_manager_sdk.zip ``` ------ Outputnya akan terlihat serupa dengan yang berikut ini: ``` Archive: aws-greengrass-stream-manager-sdk-python/stream_manager.zip Length Date Time Name --------- ---------- ----- ---- 0 02-24-2021 20:45 stream_manager/ 913 02-24-2021 20:45 stream_manager/__init__.py 9719 02-24-2021 20:45 stream_manager/utilinternal.py 1412 02-24-2021 20:45 stream_manager/exceptions.py 1004 02-24-2021 20:45 stream_manager/util.py 0 02-24-2021 20:45 stream_manager/data/ 254463 02-24-2021 20:45 stream_manager/data/__init__.py 26515 02-24-2021 20:45 stream_manager/streammanagerclient.py --------- ------- 294026 8 files ``` 1. Salin artefak SDK Stream Manager ke folder artefak komponen Anda. Selain file ZIP SDK Stream Manager, komponen Anda menggunakan `requirements.txt` untuk menginstal dependensi SDK Stream Manager. Ganti *\$1/greengrass-components* dengan path ke folder yang Anda gunakan untuk pengembangan lokal. ------ #### [ Linux or Unix ] ``` cp {stream_manager_sdk.zip,requirements.txt} ~/greengrass-components/artifacts/com.example.StreamManagerS3Python/1.0.0/ ``` ------ #### [ Windows Command Prompt (CMD) ] ``` robocopy . %USERPROFILE%\greengrass-components\artifacts\com.example.StreamManagerS3Python\1.0.0 stream_manager_sdk.zip robocopy . %USERPROFILE%\greengrass-components\artifacts\com.example.StreamManagerS3Python\1.0.0 requirements.txt ``` ------ #### [ PowerShell ] ``` cp .\stream_manager_sdk.zip,.\requirements.txt ~\greengrass-components\artifacts\com.example.StreamManagerS3Python\1.0.0\ ``` ------ 1. Buat resep komponen Anda. Dalam resep tersebut, lakukan hal berikut: 1. Tentukan `stream_manager_sdk.zip` dan `requirements.txt` sebagai artefak. 1. Tentukan aplikasi Python Anda sebagai artefak. 1. Dalam instal siklus hidup, instal persyaratan SDK Stream Manager dari `requirements.txt`. 1. Dalam jalankan siklus hidup, tambahkan SDK Stream Manager ke `PYTHONPATH`, dan jalankan aplikasi Python Anda. Resep komponen Anda mungkin terlihat serupa dengan contoh berikut ini. Komponen ini menjalankan contoh [stream\$1manager\$1s3.py](https://github.com/aws-greengrass/aws-greengrass-stream-manager-sdk-python/blob/main/samples/stream_manager_s3.py). ------ #### [ JSON ] ``` { "RecipeFormatVersion": "2020-01-25", "ComponentName": "com.example.StreamManagerS3Python", "ComponentVersion": "1.0.0", "ComponentDescription": "Uses stream manager to upload a file to an S3 bucket.", "ComponentPublisher": "Amazon", "ComponentDependencies": { "aws.greengrass.StreamManager": { "VersionRequirement": "^2.0.0" } }, "Manifests": [ { "Platform": { "os": "linux" }, "Lifecycle": { "install": "pip3 install --user -r {artifacts:path}/requirements.txt", "Run": "export PYTHONPATH=$PYTHONPATH:{artifacts:decompressedPath}/stream_manager_sdk; python3 {artifacts:path}/stream_manager_s3.py" }, "Artifacts": [ { "URI": "s3://amzn-s3-demo-bucket/artifacts/com.example.StreamManagerS3Python/1.0.0/stream_manager_sdk.zip", "Unarchive": "ZIP" }, { "URI": "s3://amzn-s3-demo-bucket/artifacts/com.example.StreamManagerS3Python/1.0.0/stream_manager_s3.py" }, { "URI": "s3://amzn-s3-demo-bucket/artifacts/com.example.StreamManagerS3Python/1.0.0/requirements.txt" } ] }, { "Platform": { "os": "windows" }, "Lifecycle": { "install": "pip3 install --user -r {artifacts:path}/requirements.txt", "Run": "set \"PYTHONPATH=%PYTHONPATH%;{artifacts:decompressedPath}/stream_manager_sdk\" & py -3 {artifacts:path}/stream_manager_s3.py" }, "Artifacts": [ { "URI": "s3://amzn-s3-demo-bucket/artifacts/com.example.StreamManagerS3Python/1.0.0/stream_manager_sdk.zip", "Unarchive": "ZIP" }, { "URI": "s3://amzn-s3-demo-bucket/artifacts/com.example.StreamManagerS3Python/1.0.0/stream_manager_s3.py" }, { "URI": "s3://amzn-s3-demo-bucket/artifacts/com.example.StreamManagerS3Python/1.0.0/requirements.txt" } ] } ] } ``` ------ #### [ YAML ] ``` --- RecipeFormatVersion: '2020-01-25' ComponentName: com.example.StreamManagerS3Python ComponentVersion: 1.0.0 ComponentDescription: Uses stream manager to upload a file to an S3 bucket. ComponentPublisher: Amazon ComponentDependencies: aws.greengrass.StreamManager: VersionRequirement: "^2.0.0" Manifests: - Platform: os: linux Lifecycle: install: pip3 install --user -r {artifacts:path}/requirements.txt Run: | export PYTHONPATH=$PYTHONPATH:{artifacts:decompressedPath}/stream_manager_sdk python3 {artifacts:path}/stream_manager_s3.py Artifacts: - URI: s3://amzn-s3-demo-bucket/artifacts/com.example.StreamManagerS3Python/1.0.0/stream_manager_sdk.zip Unarchive: ZIP - URI: s3://amzn-s3-demo-bucket/artifacts/com.example.StreamManagerS3Python/1.0.0/stream_manager_s3.py - URI: s3://amzn-s3-demo-bucket/artifacts/com.example.StreamManagerS3Python/1.0.0/requirements.txt - Platform: os: windows Lifecycle: install: pip3 install --user -r {artifacts:path}/requirements.txt Run: | set "PYTHONPATH=%PYTHONPATH%;{artifacts:decompressedPath}/stream_manager_sdk" py -3 {artifacts:path}/stream_manager_s3.py Artifacts: - URI: s3://amzn-s3-demo-bucket/artifacts/com.example.StreamManagerS3Python/1.0.0/stream_manager_sdk.zip Unarchive: ZIP - URI: s3://amzn-s3-demo-bucket/artifacts/com.example.StreamManagerS3Python/1.0.0/stream_manager_s3.py - URI: s3://amzn-s3-demo-bucket/artifacts/com.example.StreamManagerS3Python/1.0.0/requirements.txt ``` ------ Untuk informasi lebih lanjut tentang cara mengembangkan dan menguji komponen, lihat [Buat AWS IoT Greengrass komponen](create-components.md). ### Menggunakan Stream Manager SDK untuk JavaScript Stream Manager SDK untuk JavaScript tersedia sebagai kode sumber yang dapat Anda sertakan dalam komponen Anda. Buat file ZIP SDK Stream Manager, tentukan file ZIP sebagai artefak komponen, dan instal SDK dalam siklus hidup komponen. **Untuk menggunakan Stream Manager SDK untuk JavaScript** 1. Kloning atau unduh repositori [aws-greengrass-stream-manager-sdk-js](https://github.com/aws-greengrass/aws-greengrass-stream-manager-sdk-js). ``` git clone git@github.com:aws-greengrass/aws-greengrass-stream-manager-sdk-js.git ``` 1. Buat file ZIP yang berisi `aws-greengrass-stream-manager-sdk` folder, yang berisi kode sumber SDK Stream Manager untuk JavaScript. Anda dapat menyediakan file ZIP ini sebagai artefak komponen yang dibuka ritsletingnya oleh perangkat lunak AWS IoT Greengrass Core saat menginstal komponen Anda. Lakukan hal-hal berikut: 1. Buka folder yang berisi repositori yang Anda kloning atau download pada langkah sebelumnya. ``` cd aws-greengrass-stream-manager-sdk-js ``` 1. Zip folder `aws-greengrass-stream-manager-sdk` ke dalam file ZIP bernama `stream-manager-sdk.zip`. ------ #### [ Linux or Unix ] ``` zip -rv stream-manager-sdk.zip aws-greengrass-stream-manager-sdk ``` ------ #### [ Windows Command Prompt (CMD) ] ``` tar -acvf stream-manager-sdk.zip aws-greengrass-stream-manager-sdk ``` ------ #### [ PowerShell ] ``` Compress-Archive aws-greengrass-stream-manager-sdk stream-manager-sdk.zip ``` ------ 1. Verifikasi bahwa file `stream-manager-sdk.zip` berisi folder `aws-greengrass-stream-manager-sdk` dan isinya. Jalankan perintah berikut untuk menampilkan konten file ZIP. ------ #### [ Linux or Unix ] ``` unzip -l stream-manager-sdk.zip ``` ------ #### [ Windows Command Prompt (CMD) ] ``` tar -tf stream-manager-sdk.zip ``` ------ Outputnya akan terlihat serupa dengan yang berikut ini: ``` Archive: stream-manager-sdk.zip Length Date Time Name --------- ---------- ----- ---- 0 02-24-2021 22:36 aws-greengrass-stream-manager-sdk/ 369 02-24-2021 22:36 aws-greengrass-stream-manager-sdk/package.json 1017 02-24-2021 22:36 aws-greengrass-stream-manager-sdk/util.js 8374 02-24-2021 22:36 aws-greengrass-stream-manager-sdk/utilInternal.js 1937 02-24-2021 22:36 aws-greengrass-stream-manager-sdk/exceptions.js 0 02-24-2021 22:36 aws-greengrass-stream-manager-sdk/data/ 353343 02-24-2021 22:36 aws-greengrass-stream-manager-sdk/data/index.js 22599 02-24-2021 22:36 aws-greengrass-stream-manager-sdk/client.js 216 02-24-2021 22:36 aws-greengrass-stream-manager-sdk/index.js --------- ------- 387855 9 files ``` 1. Salin artefak SDK Stream Manager ke folder artefak komponen Anda. Ganti *\$1/greengrass-components* dengan path ke folder yang Anda gunakan untuk pengembangan lokal. ------ #### [ Linux or Unix ] ``` cp stream-manager-sdk.zip ~/greengrass-components/artifacts/com.example.StreamManagerS3JS/1.0.0/ ``` ------ #### [ Windows Command Prompt (CMD) ] ``` robocopy . %USERPROFILE%\greengrass-components\artifacts\com.example.StreamManagerS3JS\1.0.0 stream-manager-sdk.zip ``` ------ #### [ PowerShell ] ``` cp .\stream-manager-sdk.zip ~\greengrass-components\artifacts\com.example.StreamManagerS3JS\1.0.0\ ``` ------ 1. Buat resep komponen Anda. Dalam resep tersebut, lakukan hal berikut: 1. Tentukan `stream-manager-sdk.zip` sebagai artefak. 1. Tentukan JavaScript aplikasi Anda sebagai artefak. 1. Pada instal siklus hidup, instal SDK Manajer Pengaliran dari artefak `stream-manager-sdk.zip`. Perintah `npm install` membuat folder `node_modules` yang berisi SDK Stream Manager dan dependensinya. 1. Dalam siklus hidup run, tambahkan `node_modules` folder ke`NODE_PATH`, dan jalankan aplikasi Anda. JavaScript Resep komponen Anda mungkin terlihat serupa dengan contoh berikut ini. Komponen ini menjalankan contoh [StreamManagerS3](https://github.com/aws-greengrass/aws-greengrass-stream-manager-sdk-js/blob/main/samples/StreamManagerS3/index.js). ------ #### [ JSON ] ``` { "RecipeFormatVersion": "2020-01-25", "ComponentName": "com.example.StreamManagerS3JS", "ComponentVersion": "1.0.0", "ComponentDescription": "Uses stream manager to upload a file to an S3 bucket.", "ComponentPublisher": "Amazon", "ComponentDependencies": { "aws.greengrass.StreamManager": { "VersionRequirement": "^2.0.0" } }, "Manifests": [ { "Platform": { "os": "linux" }, "Lifecycle": { "install": "npm install {artifacts:decompressedPath}/stream-manager-sdk/aws-greengrass-stream-manager-sdk", "Run": "export NODE_PATH=$NODE_PATH:{work:path}/node_modules; node {artifacts:path}/index.js" }, "Artifacts": [ { "URI": "s3://amzn-s3-demo-bucket/artifacts/com.example.StreamManagerS3JS/1.0.0/stream-manager-sdk.zip", "Unarchive": "ZIP" }, { "URI": "s3://amzn-s3-demo-bucket/artifacts/com.example.StreamManagerS3JS/1.0.0/index.js" } ] }, { "Platform": { "os": "windows" }, "Lifecycle": { "install": "npm install {artifacts:decompressedPath}/stream-manager-sdk/aws-greengrass-stream-manager-sdk", "Run": "set \"NODE_PATH=%NODE_PATH%;{work:path}/node_modules\" & node {artifacts:path}/index.js" }, "Artifacts": [ { "URI": "s3://amzn-s3-demo-bucket/artifacts/com.example.StreamManagerS3JS/1.0.0/stream-manager-sdk.zip", "Unarchive": "ZIP" }, { "URI": "s3://amzn-s3-demo-bucket/artifacts/com.example.StreamManagerS3JS/1.0.0/index.js" } ] } ] } ``` ------ #### [ YAML ] ``` --- RecipeFormatVersion: '2020-01-25' ComponentName: com.example.StreamManagerS3JS ComponentVersion: 1.0.0 ComponentDescription: Uses stream manager to upload a file to an S3 bucket. ComponentPublisher: Amazon ComponentDependencies: aws.greengrass.StreamManager: VersionRequirement: "^2.0.0" Manifests: - Platform: os: linux Lifecycle: install: npm install {artifacts:decompressedPath}/stream-manager-sdk/aws-greengrass-stream-manager-sdk Run: | export NODE_PATH=$NODE_PATH:{work:path}/node_modules node {artifacts:path}/index.js Artifacts: - URI: s3://DOC-EXAMPLE-BUCKET/artifacts/com.example.StreamManagerS3JS/1.0.0/stream-manager-sdk.zip Unarchive: ZIP - URI: s3://DOC-EXAMPLE-BUCKET/artifacts/com.example.StreamManagerS3JS/1.0.0/index.js - Platform: os: windows Lifecycle: install: npm install {artifacts:decompressedPath}/stream-manager-sdk/aws-greengrass-stream-manager-sdk Run: | set "NODE_PATH=%NODE_PATH%;{work:path}/node_modules" node {artifacts:path}/index.js Artifacts: - URI: s3://DOC-EXAMPLE-BUCKET/artifacts/com.example.StreamManagerS3JS/1.0.0/stream-manager-sdk.zip Unarchive: ZIP - URI: s3://DOC-EXAMPLE-BUCKET/artifacts/com.example.StreamManagerS3JS/1.0.0/index.js ``` ------ Untuk informasi lebih lanjut tentang cara mengembangkan dan menguji komponen, lihat [Buat AWS IoT Greengrass komponen](create-components.md). ## Hubungkan ke manajer pengaliran dalam kode aplikasi Untuk terhubung ke stream manager dalam aplikasi Anda, buat sebuah instans `StreamManagerClient` dari SDK Stream Manager. Klien ini terhubung ke komponen stream manager pada port default 8088, atau port yang Anda tentukan. Untuk informasi lebih lanjut tentang cara menggunakan `StreamManagerClient` setelah Anda membuat instans, lihat [Gunakan StreamManagerClient untuk bekerja dengan aliran](work-with-streams.md). **Example Contoh: Hubungkan ke manajer pengaliran dengan port default** ``` import com.amazonaws.greengrass.streammanager.client.StreamManagerClient; public class MyStreamManagerComponent { void connectToStreamManagerWithDefaultPort() { StreamManagerClient client = StreamManagerClientFactory.standard().build(); // Use the client. } } ``` ``` from stream_manager import ( StreamManagerClient ) def connect_to_stream_manager_with_default_port(): client = StreamManagerClient() # Use the client. ``` ``` const { StreamManagerClient } = require('aws-greengrass-stream-manager-sdk'); function connectToStreamManagerWithDefaultPort() { const client = new StreamManagerClient(); // Use the client. } ``` **Example Contoh: Hubungkan ke manajer pengaliran dengan port non-default** Jika Anda mengonfigurasi stream manager dengan port selain default, Anda harus menggunakan [komunikasi antar proses](interprocess-communication.md) untuk mengambil port dari konfigurasi komponen. Parameter konfigurasi `port` berisi nilai yang Anda tentukan di `STREAM_MANAGER_SERVER_PORT` saat Anda menggunakan manajer pengaliran. ``` void connectToStreamManagerWithCustomPort() { EventStreamRPCConnection eventStreamRpcConnection = IPCUtils.getEventStreamRpcConnection(); GreengrassCoreIPCClient greengrassCoreIPCClient = new GreengrassCoreIPCClient(eventStreamRpcConnection); List keyPath = new ArrayList<>(); keyPath.add("port"); GetConfigurationRequest request = new GetConfigurationRequest(); request.setComponentName("aws.greengrass.StreamManager"); request.setKeyPath(keyPath); GetConfigurationResponse response = greengrassCoreIPCClient.getConfiguration(request, Optional.empty()).getResponse().get(); String port = response.getValue().get("port").toString(); System.out.print("Stream Manager is running on port: " + port); final StreamManagerClientConfig config = StreamManagerClientConfig.builder() .serverInfo(StreamManagerServerInfo.builder().port(Integer.parseInt(port)).build()).build(); StreamManagerClient client = StreamManagerClientFactory.standard().withClientConfig(config).build(); // Use the client. } ``` ``` import awsiot.greengrasscoreipc from awsiot.greengrasscoreipc.model import ( GetConfigurationRequest ) from stream_manager import ( StreamManagerClient ) TIMEOUT = 10 def connect_to_stream_manager_with_custom_port(): # Use IPC to get the port from the stream manager component configuration. ipc_client = awsiot.greengrasscoreipc.connect() request = GetConfigurationRequest() request.component_name = "aws.greengrass.StreamManager" request.key_path = ["port"] operation = ipc_client.new_get_configuration() operation.activate(request) future_response = operation.get_response() response = future_response.result(TIMEOUT) stream_manager_port = str(response.value["port"]) # Use port to create a stream manager client. stream_client = StreamManagerClient(port=stream_manager_port) # Use the client. ``` # 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](manage-data-streams.md) dan kemudian berinteraksi dengan aliran tersebut. Saat komponen membuat aliran, komponen menentukan AWS Cloud tujuan, prioritas, dan kebijakan ekspor dan penyimpanan data lainnya untuk aliran. 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](manage-data-streams.md#stream-manager-security-client-authentication). 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](#streammanagerclient-create-message-stream) + [Tambahkan pesan](#streammanagerclient-append-message) + [Baca pesan](#streammanagerclient-read-messages) + [Daftar aliran](#streammanagerclient-list-streams) + [Jelaskan aliran pesan](#streammanagerclient-describe-message-stream) + [Perbarui aliran pesan](#streammanagerclient-update-message-stream) + [Hapus aliran pesan](#streammanagerclient-delete-message-stream) ## 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 AWS IoT SiteWise Streams, dan tujuan Amazon S3 untuk ekspor otomatis. Untuk informasi selengkapnya, lihat [Ekspor konfigurasi untuk tujuan yang didukung AWS Cloud](stream-export-configurations.md). + 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 AWS IoT SiteWise , dan tujuan. 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: + [MessageStreamDefinition](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/model/MessageStreamDefinition.html)di Java SDK + [MessageStreamDefinition](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.MessageStreamDefinition.html)di Node.js SDK + [MessageStreamDefinition](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.data.html#stream_manager.data.MessageStreamDefinition)di Python SDK **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](#streammanagerclient-append-message) ke aliran tersebut untuk mengirim data untuk ekspor dan [membaca pesan](#streammanagerclient-read-messages) 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 aliran. Aliran memiliki masa pakai yang tahan lama. ### Persyaratan Komponen ini memiliki persyaratan sebagai berikut: + Versi SDK Stream Manager minimum: Python: 1.1.0  \$1  Java: 1.1.0  \$1  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\$1message\$1stream \$1 [MessageStreamDefinition](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.data.html#stream_manager.data.MessageStreamDefinition)](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.streammanagerclient.html#stream_manager.streammanagerclient.StreamManagerClient.create_message_stream) ------ #### [ 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 SDK Java: \$1 [createMessageStream[MessageStreamDefinition](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/model/MessageStreamDefinition.html)](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/client/StreamManagerClient.html#createMessageStream-com.amazonaws.greengrass.streammanager.model.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: \$1 [createMessageStream[MessageStreamDefinition](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.MessageStreamDefinition.html)](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.StreamManagerClient.html#createMessageStream) ------ Untuk informasi lebih lanjut tentang konfigurasi tujuan ekspor, lihat[Ekspor konfigurasi untuk tujuan yang didukung AWS Cloud](stream-export-configurations.md). ## 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 Manager minimum: Python: 1.1.0  \$1  Java: 1.1.0  \$1  Node.js: 1.1.0 ### Contoh #### AWS IoT Analytics atau tujuan ekspor Kinesis Data Streams Potongan berikut menambahkan pesan ke aliran bernama `StreamName`. Untuk tujuan Kinesis Data Streams AWS IoT Analytics atau Kinesis, komponen Greengrass Anda menambahkan gumpalan data. Potongan ini memiliki persyaratan sebagai berikut: + Versi SDK Stream Manager minimum: Python: 1.1.0  \$1  Java: 1.1.0  \$1  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\$1message](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.streammanagerclient.html#stream_manager.streammanagerclient.StreamManagerClient.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](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/client/StreamManagerClient.html#appendMessage-java.lang.String-byte:A-) ------ #### [ 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](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.StreamManagerClient.html#appendMessage) ------ #### AWS IoT SiteWise tujuan ekspor Potongan berikut menambahkan pesan ke aliran bernama `StreamName`. Untuk AWS IoT SiteWise tujuan, komponen Greengrass Anda menambahkan objek serial. `PutAssetPropertyValueEntry` Untuk informasi selengkapnya, lihat [Mengekspor ke AWS IoT SiteWise](stream-export-configurations.md#export-streams-to-sitewise). **catatan** Saat Anda mengirim data ke AWS IoT SiteWise, data Anda harus memenuhi persyaratan `BatchPutAssetPropertyValue` tindakan. Untuk informasi selengkapnya, lihat [BatchPutAssetPropertyValue](https://docs.aws.amazon.com/iot-sitewise/latest/APIReference/API_BatchPutAssetPropertyValue.html) di dalam *Referensi API AWS IoT SiteWise *. Potongan ini memiliki persyaratan sebagai berikut: + Versi SDK Stream Manager minimum: Python: 1.1.0  \$1  Java: 1.1.0  \$1  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\$1message \$1 [PutAssetPropertyValueEntry](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.data.html#stream_manager.data.PutAssetPropertyValueEntry)](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.streammanagerclient.html#stream_manager.streammanagerclient.StreamManagerClient.append_message) ------ #### [ 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 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](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/client/StreamManagerClient.html#appendMessage-java.lang.String-byte:A-) \$1 [PutAssetPropertyValueEntry](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/model/sitewise/PutAssetPropertyValueEntry.html) ------ #### [ 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](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.StreamManagerClient.html#appendMessage) \$1 [PutAssetPropertyValueEntry](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.PutAssetPropertyValueEntry.html) ------ #### 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](stream-export-configurations.md#export-streams-to-s3). Potongan ini memiliki persyaratan sebagai berikut: + Versi SDK Stream Manager minimum: Python: 1.1.0  \$1  Java: 1.1.0  \$1  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. ``` [https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.streammanagerclient.html#stream_manager.streammanagerclient.StreamManagerClient.append_message](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.streammanagerclient.html#stream_manager.streammanagerclient.StreamManagerClient.append_message) ------ #### [ 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 \$1](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/client/StreamManagerClient.html#appendMessage-java.lang.String-byte:A-) S3 ExportTaskDefinition](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/model/S3ExportTaskDefinition.html) ------ #### [ 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 Node.js SDK: [AppendMessage \$1](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.StreamManagerClient.html#appendMessage) S3 ExportTaskDefinition](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.S3ExportTaskDefinition.html) ------ ## Baca pesan Baca pesan dari suatu aliran. ### Persyaratan Komponen ini memiliki persyaratan sebagai berikut: + Versi SDK Stream Manager minimum: Python: 1.1.0  \$1  Java: 1.1.0  \$1  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\$1messages \$1 [ReadMessagesOptions](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.data.html#stream_manager.data.ReadMessagesOptions)](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.streammanagerclient.html#stream_manager.streammanagerclient.StreamManagerClient.read_messages) ------ #### [ Java ] ``` try (final StreamManagerClient client = StreamManagerClientFactory.standard().build()) { List 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 SDK Java: [ReadMessages \$1 [ReadMessagesOptions](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/model/ReadMessagesOptions.html)](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/client/StreamManagerClient.html#readMessages-java.lang.String-com.amazonaws.greengrass.streammanager.model.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 \$1 [ReadMessagesOptions](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.ReadMessagesOptions.html)](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.StreamManagerClient.html#readMessages) ------ ## Daftar aliran Dapatkan daftar aliran di stream manager. ### Persyaratan Komponen ini memiliki persyaratan sebagai berikut: + Versi SDK Stream Manager minimum: Python: 1.1.0  \$1  Java: 1.1.0  \$1  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\$1streams](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.streammanagerclient.html#stream_manager.streammanagerclient.StreamManagerClient.list_streams) ------ #### [ Java ] ``` try (final StreamManagerClient client = StreamManagerClientFactory.standard().build()) { List streamNames = client.listStreams(); } catch (StreamManagerException e) { // Properly handle exception. } ``` Referensi SDK Java: [listStreams](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/client/StreamManagerClient.html#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](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.StreamManagerClient.html#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 Manager minimum: Python: 1.1.0  \$1  Java: 1.1.0  \$1  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\$1message\$1stream](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.streammanagerclient.html#stream_manager.streammanagerclient.StreamManagerClient.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 Java SDK: [describeMessageStream](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/client/StreamManagerClient.html#describeMessageStream-java.lang.String-) ------ #### [ 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](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.StreamManagerClient.html#describeMessageStream) ------ ## Perbarui aliran pesan Perbarui properti dari aliran yang ada. Anda mungkin ingin memperbarui aliran jika kebutuhan Anda berubah setelah aliran dibuat. Contoh: + Tambahkan [konfigurasi ekspor](stream-export-configurations.md) baru untuk AWS Cloud tujuan. + 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.](#streammanagerclient-describe-message-stream) 1. Perbarui properti target pada `MessageStreamDefinition` dan objek bawahan yang sesuai. 1. 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 Manager minimum: Python: 1.1.0  \$1  Java: 1.1.0  \$1  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: \$1 [updateMessageStream[MessageStreamDefinition](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.data.html#stream_manager.data.MessageStreamDefinition)](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.streammanagerclient.html#stream_manager.streammanagerclient.StreamManagerClient.update_message_stream) ------ #### [ 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() {{ add(new KinesisConfig() .withIdentifier(EXPORT_IDENTIFIER) .withKinesisStreamName("test")); }}) ); } catch (StreamManagerException e) { // Properly handle exception. } ``` Referensi SDK Java: [update\$1message\$1stream](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/client/StreamManagerClient.html#updateMessageStream-java.lang.String-) \$1 [MessageStreamDefinition](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/model/MessageStreamDefinition.html) ------ #### [ 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: \$1 [updateMessageStream[MessageStreamDefinition](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.MessageStreamDefinition.html)](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.StreamManagerClient.html#updateMessageStream) ------ ### 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](#streammanagerclient-delete-message-stream) dan [buat Stream](#streammanagerclient-create-message-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](#streammanagerclient-describe-message-stream) dan kemudian periksa status penyimpanan `MessageStreamInfo` 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](#streammanagerclient-update-message-stream) nomor urutan awal konfigurasi ekspor di suatu aliran, Anda harus menentukan nilai yang kurang dari nomor urut terbaru. Untuk menemukan informasi ini, [jelaskan aliran](#streammanagerclient-describe-message-stream) dan kemudian periksa status penyimpanan `MessageStreamInfo` 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 Manager minimum: Python: 1.1.0  \$1  Java: 1.1.0  \$1  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](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.streammanagerclient.html#stream_manager.streammanagerclient.StreamManagerClient.delete_message_stream) ------ #### [ Java ] ``` try (final StreamManagerClient client = StreamManagerClientFactory.standard().build()) { client.deleteMessageStream("StreamName"); } catch (StreamManagerException e) { // Properly handle exception. } ``` Referensi SDK Java: [delete\$1message\$1stream](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/client/StreamManagerClient.html#deleteMessageStream-java.lang.String-) ------ #### [ 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](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.StreamManagerClient.html#deleteMessageStream) ------ ## Lihat juga + [Kelola aliran data di perangkat inti Greengrass](manage-data-streams.md) + [Konfigurasikan pengelola AWS IoT Greengrass aliran](configure-stream-manager.md) + [Ekspor konfigurasi untuk tujuan yang didukung AWS Cloud](stream-export-configurations.md) + `StreamManagerClient` dalam referensi SDK Stream Manager: + [Python](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.streammanagerclient.html) + [Java](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/client/StreamManagerClient.html) + [Node.js](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.StreamManagerClient.html) # Ekspor konfigurasi untuk tujuan yang didukung AWS Cloud Komponen Greengrass yang ditentukan pengguna menggunakan `StreamManagerClient` di SDK Stream Manager untuk berinteraksi dengan stream manager. Ketika komponen [membuat aliran](work-with-streams.md#streammanagerclient-create-message-stream) atau [memperbarui aliran](work-with-streams.md#streammanagerclient-create-message-stream), melewati objek `MessageStreamDefinition` yang mewakili properti aliran, termasuk definisi ekspor. Objek `ExportDefinition` berisi konfigurasi ekspor yang ditentukan untuk aliran tersebut. Stream manager menggunakan konfigurasi ekspor ini untuk menentukan di mana dan bagaimana untuk mengekspor aliran tersebut. ![\[Diagram model objek dari jenis ExportDefinition properti.\]](http://docs.aws.amazon.com/id_id/greengrass/v2/developerguide/images/stream-manager-exportconfigs.png) Anda dapat menentukan konfigurasi ekspor nol atau lebih pada suatu aliran, termasuk beberapa konfigurasi ekspor untuk satu jenis tujuan. Misalnya, Anda dapat mengekspor aliran ke dua saluran AWS IoT Analytics dan satu Kinesis data stream. Untuk upaya ekspor yang gagal, manajer aliran terus mencoba mengekspor data ke AWS Cloud interval hingga lima menit. Jumlah upaya coba lagi tidak memiliki batas maksimum. **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. **Topics** + [AWS IoT Analytics saluran](#export-to-iot-analytics) + [Amazon Kinesis data streams](#export-to-kinesis) + [AWS IoT SiteWise properti aset](#export-to-iot-sitewise) + [Objek Amazon S3](#export-to-s3) Anda bertanggung jawab untuk memelihara AWS Cloud sumber daya ini. ## AWS IoT Analytics saluran Manajer aliran mendukung ekspor otomatis ke AWS IoT Analytics. AWS IoT Analytics memungkinkan Anda melakukan analisis lanjutan pada data Anda untuk membantu membuat keputusan bisnis dan meningkatkan model pembelajaran mesin. Untuk informasi lebih lanjut, lihat [Apa itu AWS IoT Analytics?](https://docs.aws.amazon.com/iotanalytics/latest/userguide/welcome.html) dalam *AWS IoT Analytics User Guide*. Dalam SDK Stream Manager, komponen Greengrass Anda menggunakan `IoTAnalyticsConfig` untuk menentukan konfigurasi ekspor untuk jenis tujuan ini. Untuk informasi lebih lanjut, lihat referensi SDK untuk bahasa target Anda: + [TAnalyticsKonfigurasi Io di SDK](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.data.html#stream_manager.data.IoTAnalyticsConfig) Python + [Io TAnalytics Config](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/model/export/IoTAnalyticsConfig.html) di Java SDK + [Io TAnalytics Config di Node.js](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.IoTAnalyticsConfig.html) SDK ### Persyaratan Tujuan ekspor ini memiliki persyaratan sebagai berikut: + Saluran target AWS IoT Analytics harus sama Akun AWS dan Wilayah AWS sebagai perangkat inti Greengrass. + [Otorisasi perangkat inti untuk berinteraksi dengan layanan AWS](device-service-role.md) harus memungkinkan izin `iotanalytics:BatchPutMessage` untuk menargetkan saluran. Contoh: ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Effect": "Allow", "Action": [ "iotanalytics:BatchPutMessage" ], "Resource": [ "arn:aws:iotanalytics:us-east-1:123456789012:channel/channel_1_name", "arn:aws:iotanalytics:us-east-1:123456789012:channel/channel_2_name" ] } ] } ``` ------ Anda dapat memberikan akses terperinci atau bersyarat ke sumber daya, misalnya dengan menggunakan skema penamaan wildcard `*`. Untuk informasi lebih lanjut, lihat [Menambahkan dan menghapus kebijakan IAM](https://docs.aws.amazon.com/IAM/latest/UserGuide/access_policies_manage-attach-detach.html) dalam *Panduan Pengguna IAM*. ### Mengekspor ke AWS IoT Analytics Untuk membuat aliran yang diekspor AWS IoT Analytics, [komponen Greengrass Anda membuat aliran dengan definisi ekspor yang menyertakan satu](work-with-streams.md#streammanagerclient-create-message-stream) atau beberapa objek. `IoTAnalyticsConfig` Objek ini menentukan pengaturan ekspor, seperti target saluran, ukuran batch, interval batch, dan prioritas. Ketika komponen Greengrass Anda menerima data dari perangkat, komponen itu [menambahkan pesan](work-with-streams.md#streammanagerclient-append-message) yang berisi gumpalan data ke aliran target. Kemudian, stream manager mengekspor data berdasarkan pengaturan batch dan prioritas yang ditentukan dalam konfigurasi ekspor aliran. ## Amazon Kinesis data streams Stream manager mendukung ekspor otomatis ke Amazon Kinesis Data Streams. Kinesis Data Streams biasanya digunakan untuk mengumpulkan data volume tinggi dan memuatnya ke dalam gudang data atau cluster. MapReduce Untuk informasi selengkapnya, lihat [Apa yang dimaksud dengan Amazon Kinesis Data Streams?](https://docs.aws.amazon.com/streams/latest/dev/what-is-this-service.html) dalam *Panduan Developer Amazon Kinesis*. Dalam SDK Stream Manager, komponen Greengrass Anda menggunakan `KinesisConfig` untuk menentukan konfigurasi ekspor untuk jenis tujuan ini. Untuk informasi lebih lanjut, lihat referensi SDK untuk bahasa target Anda: + [KinesisConfig](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.data.html#stream_manager.data.KinesisConfig)di Python SDK + [KinesisConfig](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/model/export/KinesisConfig.html)di Java SDK + [KinesisConfig](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.KinesisConfig.html)di Node.js SDK ### Persyaratan Tujuan ekspor ini memiliki persyaratan sebagai berikut: + Aliran target di Kinesis Data Streams harus sama Akun AWS dan Wilayah AWS sebagai perangkat inti Greengrass. + (Disarankan) Stream manager v2.2.1 meningkatkan kinerja mengekspor stream ke tujuan Kinesis Data Streams. [Untuk menggunakan peningkatan dalam versi terbaru ini, tingkatkan [komponen pengelola aliran](stream-manager-component.md) Anda ke v2.2.1 dan gunakan `kinesis:ListShards` kebijakan dalam peran pertukaran token Greengrass.](device-service-role.md) + [Otorisasi perangkat inti untuk berinteraksi dengan layanan AWS](device-service-role.md) harus memungkinkan izin `kinesis:PutRecords` untuk menargetkan aliran data. Contoh: ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Effect": "Allow", "Action": [ "kinesis:PutRecords" ], "Resource": [ "arn:aws:kinesis:us-east-1:123456789012:stream/stream_1_name", "arn:aws:kinesis:us-east-1:123456789012:stream/stream_2_name" ] } ] } ``` ------ Anda dapat memberikan akses terperinci atau bersyarat ke sumber daya, misalnya dengan menggunakan skema penamaan wildcard `*`. Untuk informasi lebih lanjut, lihat [Menambahkan dan menghapus kebijakan IAM](https://docs.aws.amazon.com/IAM/latest/UserGuide/access_policies_manage-attach-detach.html) dalam *Panduan Pengguna IAM*. ### Mengekspor ke Kinesis Data Streams Untuk membuat pengaliran yang diekspor ke Kinesis Data Streams, komponen Greengrass Anda [membuat pengaliran](work-with-streams.md#streammanagerclient-create-message-stream) dengan definisi ekspor yang mencakup satu atau lebih objek `KinesisConfig`. Objek ini menentukan pengaturan ekspor, seperti target aliran data, ukuran batch, interval batch, dan prioritas. Ketika komponen Greengrass Anda menerima data dari perangkat, komponen itu [menambahkan pesan](work-with-streams.md#streammanagerclient-append-message) yang berisi gumpalan data ke aliran target. Kemudian, stream manager mengekspor data berdasarkan pengaturan batch dan prioritas yang ditentukan dalam konfigurasi ekspor aliran. Stream manager menghasilkan UUID acak yang unik sebagai kunci partisi untuk setiap rekaman yang diunggah ke Amazon Kinesis. ## AWS IoT SiteWise properti aset Manajer aliran mendukung ekspor otomatis ke AWS IoT SiteWise. AWS IoT SiteWise memungkinkan Anda mengumpulkan, mengatur, dan menganalisis data dari peralatan industri dalam skala besar. Untuk informasi lebih lanjut, lihat [Apa itu AWS IoT SiteWise?](https://docs.aws.amazon.com/iot-sitewise/latest/userguide/what-is-sitewise.html) dalam *AWS IoT SiteWise User Guide*. Dalam SDK Stream Manager, komponen Greengrass Anda menggunakan `IoTSiteWiseConfig` untuk menentukan konfigurasi ekspor untuk jenis tujuan ini. Untuk informasi lebih lanjut, lihat referensi SDK untuk bahasa target Anda: + [Io TSite WiseConfig](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.data.html#stream_manager.data.IoTSiteWiseConfig) dalam SDK Python + [Io TSite WiseConfig](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/model/export/IoTSiteWiseConfig.html) di Java SDK + [Io TSite WiseConfig](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.IoTSiteWiseConfig.html) di Node.js SDK **catatan** AWS juga menyediakan AWS IoT SiteWise komponen, yang menawarkan solusi pra-bangun yang dapat Anda gunakan untuk mengalirkan data dari sumber OPC-UA. Untuk informasi selengkapnya, lihat [Kolektor IoT SiteWise OPC UA](iotsitewise-opcua-collector-component.md). ### Persyaratan Tujuan ekspor ini memiliki persyaratan sebagai berikut: + Properti aset target AWS IoT SiteWise harus sama Akun AWS dan Wilayah AWS sebagai perangkat inti Greengrass. **catatan** Untuk daftar Wilayah AWS s yang AWS IoT SiteWise mendukung, lihat [AWS IoT SiteWise titik akhir dan kuota](https://docs.aws.amazon.com/general/latest/gr/iot-sitewise.html#iot-sitewise_region) di Referensi *AWS Umum*. + [Otorisasi perangkat inti untuk berinteraksi dengan layanan AWS](device-service-role.md) harus memungkinkan izin `iotsitewise:BatchPutAssetPropertyValue` untuk menargetkan properti aset. Kebijakan berikut menggunakan kunci kondisi `iotsitewise:assetHierarchyPath` untuk memberikan akses ke aset akar target dan anak-anaknya. Anda dapat menghapus `Condition` dari kebijakan untuk mengizinkan akses ke semua AWS IoT SiteWise aset Anda atau menentukan ARNs aset individu. ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Effect": "Allow", "Action": "iotsitewise:BatchPutAssetPropertyValue", "Resource": "*", "Condition": { "StringLike": { "iotsitewise:assetHierarchyPath": [ "/root node asset ID", "/root node asset ID/*" ] } } } ] } ``` ------ Anda dapat memberikan akses terperinci atau bersyarat ke sumber daya, misalnya dengan menggunakan skema penamaan wildcard `*`. Untuk informasi lebih lanjut, lihat [Menambahkan dan menghapus kebijakan IAM](https://docs.aws.amazon.com/IAM/latest/UserGuide/access_policies_manage-attach-detach.html) dalam *Panduan Pengguna IAM*. Untuk informasi keamanan penting, lihat [ BatchPutAssetPropertyValue otorisasi](https://docs.aws.amazon.com/iot-sitewise/latest/userguide/security_iam_service-with-iam.html#security_iam_service-with-iam-id-based-policies-batchputassetpropertyvalue-action) di *Panduan AWS IoT SiteWise Pengguna*. ### Mengekspor ke AWS IoT SiteWise Untuk membuat aliran yang diekspor AWS IoT SiteWise, [komponen Greengrass Anda membuat aliran dengan definisi ekspor yang menyertakan satu](work-with-streams.md#streammanagerclient-create-message-stream) atau beberapa objek. `IoTSiteWiseConfig` Objek ini menentukan pengaturan ekspor, seperti ukuran batch, interval batch, dan prioritas. Ketika komponen Greengrass Anda menerima data properti aset dari perangkat, komponen itu menambahkan pesan yang berisi data ke aliran target. Pesan merupakan objek `PutAssetPropertyValueEntry` berseri JSON yang berisi nilai properti untuk satu atau lebih properti aset. Untuk informasi selengkapnya, lihat: [Tambahkan pesan](work-with-streams.md#streammanagerclient-append-message-sitewise) untuk tujuan ekspor AWS IoT SiteWise . **catatan** Saat Anda mengirim data ke AWS IoT SiteWise, data Anda harus memenuhi persyaratan `BatchPutAssetPropertyValue` tindakan. Untuk informasi selengkapnya, lihat [BatchPutAssetPropertyValue](https://docs.aws.amazon.com/iot-sitewise/latest/APIReference/API_BatchPutAssetPropertyValue.html) di dalam *Referensi API AWS IoT SiteWise *. Kemudian, stream manager mengekspor data berdasarkan pengaturan batch dan prioritas yang ditentukan dalam konfigurasi ekspor aliran. Anda dapat menyesuaikan pengaturan stream manager dan logika komponen Greengrass untuk merancang strategi ekspor Anda. Contoh: + Untuk ekspor yang mendekati real time, atur ukuran batch dan pengaturan interval yang rendah dan tambahkan data ke aliran saat diterima. + Untuk mengoptimalkan batching, mengurangi batasan bandwidth, atau meminimalkan biaya, komponen Greengrass Anda dapat mengumpulkan titik data timestamp-quality-value (TQV) yang diterima untuk satu properti aset sebelum menambahkan data ke aliran. Salah satu strateginya adalah mengelompokkan entri hingga 10 kombinasi aset properti yang berbeda, atau alias properti, dalam satu pesan, dan bukan mengirim lebih dari satu entri untuk properti yang sama. Hal ini membantu manajer pengaliran untuk tetap berada dalam [kuota AWS IoT SiteWise](https://docs.aws.amazon.com/iot-sitewise/latest/userguide/quotas.html). ## Objek Amazon S3 Stream manager mendukung ekspor otomatis ke Amazon S3. Sebagai contoh, Anda dapat menggunakan Amazon S3 untuk menyimpan dan mengambil sejumlah besar data. Untuk informasi selengkapnya, lihat [Apa itu Amazon S3?](https://docs.aws.amazon.com/AmazonS3/latest/dev/Welcome.html) dalam *Panduan Developer Amazon Simple Storage Service*. Dalam SDK Stream Manager, komponen Greengrass Anda menggunakan `S3ExportTaskExecutorConfig` untuk menentukan konfigurasi ekspor untuk jenis tujuan ini. Untuk informasi lebih lanjut, lihat referensi SDK untuk bahasa target Anda: + [S3 ExportTaskExecutorConfig](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.data.html#stream_manager.data.S3ExportTaskExecutorConfig) di Python SDK + [S3 ExportTaskExecutorConfig](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/model/export/S3ExportTaskExecutorConfig.html) di Java SDK + [S3 ExportTaskExecutorConfig](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.S3ExportTaskExecutorConfig.html) di Node.js SDK ### Persyaratan Tujuan ekspor ini memiliki persyaratan sebagai berikut: + Target ember Amazon S3 harus sama dengan perangkat inti Akun AWS Greengrass. + Jika fungsi Lambda yang berjalan dalam mode **wadah Greengrass** menulis file input ke direktori file input, Anda harus memasang direktori sebagai volume dalam wadah dengan izin tulis. Ini memastikan bahwa file ditulis ke sistem file root dan terlihat oleh komponen manajer aliran, yang berjalan di luar wadah. + Jika komponen kontainer Docker menulis file input ke direktori file input, Anda harus me-mount direktori sebagai volume dalam wadah dengan izin tulis. Ini memastikan bahwa file ditulis ke sistem file root dan terlihat oleh komponen manajer aliran, yang berjalan di luar wadah. + [Otorisasi perangkat inti untuk berinteraksi dengan layanan AWS](device-service-role.md) harus memungkinkan izin berikut untuk target bucket. Contoh: ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Effect": "Allow", "Action": [ "s3:PutObject", "s3:AbortMultipartUpload", "s3:ListMultipartUploadParts" ], "Resource": [ "arn:aws:s3:::bucket-1-name/*", "arn:aws:s3:::bucket-2-name/*" ] } ] } ``` ------ Anda dapat memberikan akses terperinci atau bersyarat ke sumber daya, misalnya dengan menggunakan skema penamaan wildcard `*`. Untuk informasi lebih lanjut, lihat [Menambahkan dan menghapus kebijakan IAM](https://docs.aws.amazon.com/IAM/latest/UserGuide/access_policies_manage-attach-detach.html) dalam *Panduan Pengguna IAM*. ### Mengekspor ke Amazon S3 Untuk membuat aliran yang diekspor ke Amazon S3, komponen Greengrass Anda menggunakan objek `S3ExportTaskExecutorConfig` untuk mengonfigurasi kebijakan ekspor. Kebijakan ini menentukan pengaturan ekspor, seperti ambang batas dan prioritas unggahan multipart. Untuk ekspor Amazon S3, stream manager akan mengunggah data yang dibaca dari file lokal pada perangkat inti. Untuk memulai unggahan, komponen Greengrass Anda menambahkan tugas ekspor ke aliran target. Tugas ekspor berisi informasi tentang file input dan objek Amazon S3 target. Stream manager menjalankan tugas dalam urutan yang ditambahkan ke aliran. **catatan** Ember target harus sudah ada di Anda Akun AWS. Jika objek untuk kunci tertentu tidak ada, stream manager akan membuatnya untuk Anda. Manajer pengaliran menggunakan properti ambang batas unggahan multipart, pengaturan [ukuran bagian minimum](configure-stream-manager.md#stream-manager-minimum-part-size), dan ukuran file input untuk menentukan cara mengunggah data. Ambang batas unggahan multipart harus lebih besar atau sama dengan ukuran bagian minimum. Jika Anda ingin meng-upload data secara paralel, Anda dapat membuat beberapa aliran. Kunci yang menentukan objek Amazon S3 target Anda dapat menyertakan DateTimeFormatter string [Java](https://docs.oracle.com/javase/8/docs/api/java/time/format/DateTimeFormatter.html) yang valid di placeholder. `!{timestamp:value}` Anda dapat menggunakan placeholder stempel waktu ini untuk membagi data di Amazon S3 berdasarkan waktu data file input tersebut diunggah. Sebagai contoh, nama kunci berikut memutuskan untuk nilai seperti `my-key/2020/12/31/data.txt`. ``` my-key/!{timestamp:YYYY}/!{timestamp:MM}/!{timestamp:dd}/data.txt ``` **catatan** Jika Anda ingin memantau status ekspor untuk suatu aliran, pertama-tama buat status aliran dan kemudian konfigurasi aliran ekspor untuk menggunakannya. Untuk informasi selengkapnya, lihat [Pantau tugas ekspor](#monitor-export-status-s3). #### Kelola data input Anda dapat menuliskan kode yang digunakan oleh aplikasi IoT untuk mengelola siklus hidup data input. Contoh alur kerja berikut menunjukkan bagaimana Anda mungkin menggunakan komponen Greengrass untuk mengelola data ini. 1. Proses lokal menerima data dari perangkat atau periferal, dan kemudian menuliskan data ke file dalam direktori pada perangkat inti. Ini adalah file input untuk stream manager. 1. Komponen Greengrass memindai direktori dan [menambahkan tugas ekspor](work-with-streams.md#streammanagerclient-append-message-export-task) ke aliran target ketika file baru dibuat. Tugas tersebut adalah objek `S3ExportTaskDefinition` JSON berseri yang menentukan URL file input, bucket dan kunci Amazon S3 target, dan metadata pengguna opsional. 1. Stream manager membaca file input dan mengekspor data ke Amazon S3 dalam urutan tugas yang ditambahkan. Ember target harus sudah ada di Anda Akun AWS. Jika objek untuk kunci tertentu tidak ada, stream manager akan membuatnya untuk Anda. 1. Komponen Greengrass [membaca pesan](work-with-streams.md#streammanagerclient-read-messages) dari aliran status untuk memantau status ekspor. Setelah tugas ekspor selesai, komponen Greengrass dapat menghapus file input yang sesuai. Untuk informasi selengkapnya, lihat [Pantau tugas ekspor](#monitor-export-status-s3). ### Pantau tugas ekspor Anda dapat menuliskan kode yang digunakan oleh aplikasi IoT untuk memantau status ekspor Amazon S3 Anda. Komponen Greengrass Anda harus membuat status aliran dan kemudian mengonfigurasi aliran ekspor untuk menuliskan pembaruan status pada status aliran. Aliran status tunggal dapat menerima pembaruan status dari beberapa aliran yang diekspor ke Amazon S3. Pertama-tama, [buat aliran](work-with-streams.md#streammanagerclient-create-message-stream) yang akan digunakan sebagai status aliran. Anda dapat mengonfigurasi kebijakan ukuran dan penyimpanan bagi aliran tersebut untuk mengontrol umur pesan status. Contoh: + Tetapkan `Persistence` ke `Memory` jika Anda tidak ingin menyimpan pesan status. + Tetapkan `StrategyOnFull` ke `OverwriteOldestData` agar pesan status baru tidak hilang. Kemudian, buat atau perbarui aliran ekspor untuk menggunakan status aliran. Secara khusus, atur properti konfigurasi status pada konfigurasi ekspor `S3ExportTaskExecutorConfig` aliran. Pengaturan ini memberitahu stream manager untuk menuliskan pesan status tentang tugas ekspor ke status aliran. Di objek `StatusConfig`, tentukan nama status pengaliran dan tingkat verbositasnya. Nilai yang didukung berikut berkisar dari verbositas paling kecil (`ERROR`) hingga verbositas tertinggi (`TRACE`). Default-nya adalah `INFO`. + `ERROR` + `WARN` + `INFO` + `DEBUG` + `TRACE` Contoh alur kerja berikut menunjukkan bagaimana komponen Greengrass mungkin menggunakan status aliran untuk memantau status ekspor. 1. Seperti yang dijelaskan dalam alur kerja sebelumnya, sebuah komponen Greengrass [menambahkan tugas ekspor](work-with-streams.md#streammanagerclient-append-message-export-task) ke aliran yang dikonfigurasi untuk menuliskan pesan status tentang tugas ekspor ke aliran status. Operasi append mengembalikan nomor urut yang mewakili ID tugas. 1. Komponen Greengrass [membaca pesan](work-with-streams.md#streammanagerclient-read-messages) secara berurutan dari aliran status, dan kemudian menyaring pesan berdasarkan nama aliran dan ID tugas atau berdasarkan properti tugas ekspor dari konteks pesan. Sebagai contoh, komponen Greengrass dapat memfilter dengan URL file input dari tugas ekspor, yang diwakili oleh objek `S3ExportTaskDefinition` dalam konteks pesan tersebut. Kode status berikut menunjukkan bahwa tugas ekspor telah mencapai keadaan lengkap: + `Success`. Upload berhasil diselesaikan. + `Failure`. Stream manager mengalami kesalahan, misalnya, bucket yang ditentukan tidak ada. Setelah menyelesaikan masalah, Anda dapat menambahkan tugas ekspor ke aliran lagi. + `Canceled`. Tugas dihentikan karena definisi aliran atau ekspor dihapus, atau periode time-to-live (TTL) tugas berakhir. **catatan** Tugas mungkin juga memiliki status `InProgress` atau `Warning`. Stream manager mengeluarkan peringatan ketika peristiwa mengembalikan kesalahan yang tidak mempengaruhi pelaksanaan tugas. Sebagai contoh, kegagalan untuk membersihkan sebagian unggahan akan mengembalikan peringatan. 1. Setelah tugas ekspor selesai, komponen Greengrass dapat menghapus file input yang sesuai. Contoh berikut menunjukkan bagaimana komponen Greengrass mungkin membaca dan memproses pesan status. ------ #### [ Python ] ``` import time from stream_manager import ( ReadMessagesOptions, Status, StatusConfig, StatusLevel, StatusMessage, StreamManagerClient, ) from stream_manager.util import Util client = StreamManagerClient() try: # Read the statuses from the export status stream is_file_uploaded_to_s3 = False while not is_file_uploaded_to_s3: try: messages_list = client.read_messages( "StatusStreamName", ReadMessagesOptions(min_message_count=1, read_timeout_millis=1000) ) for message in messages_list: # Deserialize the status message first. status_message = Util.deserialize_json_bytes_to_obj(message.payload, StatusMessage) # Check the status of the status message. If the status is "Success", # the file was successfully uploaded to S3. # If the status was either "Failure" or "Cancelled", the server was unable to upload the file to S3. # We will print the message for why the upload to S3 failed from the status message. # If the status was "InProgress", the status indicates that the server has started uploading # the S3 task. if status_message.status == Status.Success: logger.info("Successfully uploaded file at path " + file_url + " to S3.") is_file_uploaded_to_s3 = True elif status_message.status == Status.Failure or status_message.status == Status.Canceled: logger.info( "Unable to upload file at path " + file_url + " to S3. Message: " + status_message.message ) is_file_uploaded_to_s3 = True time.sleep(5) except StreamManagerException: logger.exception("Exception while running") except StreamManagerException: pass # Properly handle errors. except ConnectionError or asyncio.TimeoutError: pass # Properly handle errors. ``` [Referensi Python SDK: read\$1messages \$1 [StatusMessage](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.data.html#stream_manager.data.StatusMessage)](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.streammanagerclient.html#stream_manager.streammanagerclient.StreamManagerClient.read_messages) ------ #### [ Java ] ``` import com.amazonaws.greengrass.streammanager.client.StreamManagerClient; import com.amazonaws.greengrass.streammanager.client.StreamManagerClientFactory; import com.amazonaws.greengrass.streammanager.client.utils.ValidateAndSerialize; import com.amazonaws.greengrass.streammanager.model.ReadMessagesOptions; import com.amazonaws.greengrass.streammanager.model.Status; import com.amazonaws.greengrass.streammanager.model.StatusConfig; import com.amazonaws.greengrass.streammanager.model.StatusLevel; import com.amazonaws.greengrass.streammanager.model.StatusMessage; try (final StreamManagerClient client = StreamManagerClientFactory.standard().build()) { try { boolean isS3UploadComplete = false; while (!isS3UploadComplete) { try { // Read the statuses from the export status stream List messages = client.readMessages("StatusStreamName", new ReadMessagesOptions().withMinMessageCount(1L).withReadTimeoutMillis(1000L)); for (Message message : messages) { // Deserialize the status message first. StatusMessage statusMessage = ValidateAndSerialize.deserializeJsonBytesToObj(message.getPayload(), StatusMessage.class); // Check the status of the status message. If the status is "Success", the file was successfully uploaded to S3. // If the status was either "Failure" or "Canceled", the server was unable to upload the file to S3. // We will print the message for why the upload to S3 failed from the status message. // If the status was "InProgress", the status indicates that the server has started uploading the S3 task. if (Status.Success.equals(statusMessage.getStatus())) { System.out.println("Successfully uploaded file at path " + FILE_URL + " to S3."); isS3UploadComplete = true; } else if (Status.Failure.equals(statusMessage.getStatus()) || Status.Canceled.equals(statusMessage.getStatus())) { System.out.println(String.format("Unable to upload file at path %s to S3. Message %s", statusMessage.getStatusContext().getS3ExportTaskDefinition().getInputUrl(), statusMessage.getMessage())); sS3UploadComplete = true; } } } catch (StreamManagerException ignored) { } finally { // Sleep for sometime for the S3 upload task to complete before trying to read the status message. Thread.sleep(5000); } } catch (e) { // Properly handle errors. } } catch (StreamManagerException e) { // Properly handle exception. } ``` Referensi SDK Java: [ReadMessages \$1 [StatusMessage](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/model/StatusMessage.html)](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/client/StreamManagerClient.html#readMessages-java.lang.String-com.amazonaws.greengrass.streammanager.model.ReadMessagesOptions-) ------ #### [ Node.js ] ``` const { StreamManagerClient, ReadMessagesOptions, Status, StatusConfig, StatusLevel, StatusMessage, util, } = require(*'aws-greengrass-stream-manager-sdk'*); const client = new StreamManagerClient(); client.onConnected(async () => { try { let isS3UploadComplete = false; while (!isS3UploadComplete) { try { // Read the statuses from the export status stream const messages = await c.readMessages("StatusStreamName", new ReadMessagesOptions() .withMinMessageCount(1) .withReadTimeoutMillis(1000)); messages.forEach((message) => { // Deserialize the status message first. const statusMessage = util.deserializeJsonBytesToObj(message.payload, StatusMessage); // Check the status of the status message. If the status is 'Success', the file was successfully uploaded to S3. // If the status was either 'Failure' or 'Cancelled', the server was unable to upload the file to S3. // We will print the message for why the upload to S3 failed from the status message. // If the status was "InProgress", the status indicates that the server has started uploading the S3 task. if (statusMessage.status === Status.Success) { console.log(`Successfully uploaded file at path ${FILE_URL} to S3.`); isS3UploadComplete = true; } else if (statusMessage.status === Status.Failure || statusMessage.status === Status.Canceled) { console.log(`Unable to upload file at path ${FILE_URL} to S3. Message: ${statusMessage.message}`); isS3UploadComplete = true; } }); // Sleep for sometime for the S3 upload task to complete before trying to read the status message. await new Promise((r) => setTimeout(r, 5000)); } catch (e) { // Ignored } } 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 \$1 [StatusMessage](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.StatusMessage.html)](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.StreamManagerClient.html#readMessages) ------