Terjemahan disediakan oleh mesin penerjemah. Jika konten terjemahan yang diberikan bertentangan dengan versi bahasa Inggris aslinya, utamakan versi bahasa Inggris. # Ikhtisar fitur pipeline di Amazon OpenSearch Ingestion Amazon OpenSearch Ingestion menyediakan *saluran pipa*, yang terdiri dari sumber, buffer, nol atau lebih prosesor, dan satu atau lebih sink. Saluran pipa konsumsi ditenagai oleh Data Prepper sebagai mesin data. Untuk ikhtisar berbagai komponen pipa, lihat[Konsep kunci dalam Amazon OpenSearch Ingestion](ingestion-process.md). Bagian berikut memberikan gambaran umum tentang beberapa fitur yang paling umum digunakan di Amazon OpenSearch Ingestion. **catatan** Ini bukan daftar lengkap fitur yang tersedia untuk saluran pipa. Untuk dokumentasi komprehensif dari semua fungsi pipeline yang tersedia, lihat [dokumentasi Data Prepper](https://opensearch.org/docs/latest/data-prepper/pipelines/pipelines/). Perhatikan bahwa OpenSearch Ingestion menempatkan beberapa kendala pada plugin dan opsi yang dapat Anda gunakan. Untuk informasi selengkapnya, lihat [Plugin dan opsi yang didukung untuk saluran Amazon OpenSearch Ingestion](pipeline-config-reference.md). **Topics** + [Buffering persisten](#persistent-buffering) + [Memisahkan](#osis-features-splitting) + [Mengikat](#osis-features-chaining) + [Antrean dead-letter](#osis-features-dlq) + [Manajemen indeks](#osis-features-index-management) + [End-to-end pengakuan](#osis-features-e2e) + [Sumber tekanan balik](#osis-features-backpressure) ## Buffering persisten Buffer persisten menyimpan data Anda dalam buffer berbasis disk di beberapa Availability Zone untuk meningkatkan daya tahan data. Anda dapat menggunakan buffering persisten untuk menyerap data dari semua sumber berbasis push yang didukung tanpa menyiapkan buffer mandiri. Sumber-sumber ini termasuk HTTP dan OpenTelemetry untuk log, jejak, dan metrik. Untuk mengaktifkan buffering persisten, pilih **Aktifkan buffer persisten** saat Anda membuat atau memperbarui pipeline. Untuk informasi selengkapnya, lihat [Membuat saluran pipa Amazon OpenSearch Ingestion](creating-pipeline.md). OpenSearch Ingestion secara dinamis menentukan jumlah yang akan digunakan OCUs untuk buffering persisten, anjak piutang dalam sumber data, transformasi streaming, dan tujuan sink. Karena mengalokasikan beberapa OCUs untuk buffering, Anda mungkin perlu meningkatkan minimum dan maksimum OCUs untuk mempertahankan throughput konsumsi yang sama. Pipa menyimpan data dalam buffer hingga 72 jam. Jika Anda mengaktifkan buffering persisten untuk pipeline, ukuran payload permintaan maksimum default adalah sebagai berikut: + **Sumber HTTP** - 10 MB + **OpenTelemetry sumber** - 4 MB Untuk sumber HTTP, Anda dapat meningkatkan ukuran payload maksimum hingga 20 MB. Ukuran payload permintaan mencakup seluruh permintaan HTTP, yang biasanya berisi beberapa peristiwa. Setiap acara tidak boleh melebihi 3,5 MB. Pipa dengan buffering persisten membagi unit pipa yang dikonfigurasi antara unit komputasi dan buffer. Jika pipeline menggunakan prosesor intensif CPU seperti grok, key-value, atau split string, ia mengalokasikan unit dalam rasio 1:1. buffer-to-compute Jika tidak, itu mengalokasikannya dalam rasio 3:1, selalu mendukung unit komputasi. Contoh: + Pipa dengan grok dan 2 unit maks — 1 unit komputasi dan 1 unit penyangga + Pipa dengan grok dan 5 unit maks - 3 unit komputasi dan 2 unit penyangga + Pipeline tanpa prosesor dan 2 unit maks - 1 unit komputasi dan 1 unit buffer + Pipa tanpa prosesor dan 4 unit maks - 1 unit komputasi dan 3 unit penyangga + Pipa dengan grok dan 5 unit maks - 2 unit komputasi dan 3 unit penyangga Secara default, pipeline menggunakan file Kunci milik AWS untuk mengenkripsi data buffer. Pipeline ini tidak memerlukan izin tambahan untuk peran pipeline. Sebagai alternatif, Anda dapat menentukan kunci terkelola pelanggan dan menambahkan izin IAM berikut ke peran pipeline: ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Sid": "KeyAccess", "Effect": "Allow", "Action": [ "kms:Decrypt", "kms:GenerateDataKeyWithoutPlaintext" ], "Resource": "arn:aws:kms:us-east-1:111122223333:key/ASIAIOSFODNN7EXAMPLE" } ] } ``` ------ Untuk informasi selengkapnya, lihat [Kunci terkelola pelanggan](https://docs.aws.amazon.com/kms/latest/developerguide/concepts.html#customer-cmk) di *Panduan AWS Key Management Service Pengembang*. **catatan** Jika Anda menonaktifkan buffering persisten, pipeline Anda mulai berjalan sepenuhnya pada buffering dalam memori. ## Memisahkan Anda dapat mengonfigurasi pipeline OpenSearch Ingestion untuk *membagi* peristiwa masuk menjadi sub-pipeline, memungkinkan Anda melakukan berbagai jenis pemrosesan pada acara masuk yang sama. Contoh pipeline berikut membagi peristiwa yang masuk menjadi dua sub-pipeline. Setiap sub-pipeline menggunakan prosesornya sendiri untuk memperkaya dan memanipulasi data, dan kemudian mengirimkan data ke indeks yang berbeda. OpenSearch ``` version: "2" log-pipeline: source: http: ... sink: - pipeline: name: "logs_enriched_one_pipeline" - pipeline: name: "logs_enriched_two_pipeline" logs_enriched_one_pipeline: source: pipeline: name: "log-pipeline" processor: ... sink: - opensearch: # Provide a domain or collection endpoint # Enable the 'serverless' flag if the sink is an OpenSearch Serverless collection aws: ... index: "enriched_one_logs" logs_enriched_two_pipeline: source: pipeline: name: "log-pipeline" processor: ... sink: - opensearch: # Provide a domain or collection endpoint # Enable the 'serverless' flag if the sink is an OpenSearch Serverless collection aws: ... index: "enriched_two_logs" ``` ## Mengikat Anda dapat *menghubungkan* beberapa sub-pipeline bersama-sama untuk melakukan pemrosesan dan pengayaan data dalam potongan. Dengan kata lain, Anda dapat memperkaya acara yang masuk dengan kemampuan pemrosesan tertentu dalam satu sub-pipa, kemudian mengirimkannya ke sub-pipa lain untuk pengayaan tambahan dengan prosesor yang berbeda, dan akhirnya mengirimkannya ke wastafelnya. OpenSearch Dalam contoh berikut, `log_pipeline` sub-pipeline memperkaya peristiwa log masuk dengan satu set prosesor, kemudian mengirimkan acara ke indeks bernama. OpenSearch `enriched_logs` Pipeline mengirimkan peristiwa yang sama ke `log_advanced_pipeline` sub-pipeline, yang memprosesnya dan mengirimkannya ke OpenSearch indeks yang berbeda bernama`enriched_advanced_logs`. ``` version: "2" log-pipeline: source: http: ... processor: ... sink: - opensearch: # Provide a domain or collection endpoint # Enable the 'serverless' flag if the sink is an OpenSearch Serverless collection aws: ... index: "enriched_logs" - pipeline: name: "log_advanced_pipeline" log_advanced_pipeline: source: pipeline: name: "log-pipeline" processor: ... sink: - opensearch: # Provide a domain or collection endpoint # Enable the 'serverless' flag if the sink is an OpenSearch Serverless collection aws: ... index: "enriched_advanced_logs" ``` ## Antrean dead-letter Antrian huruf mati (DLQs) adalah tujuan untuk peristiwa yang gagal ditulis oleh pipa ke wastafel. Di OpenSearch Ingestion, Anda harus menentukan bucket Amazon S3 dengan izin tulis yang sesuai untuk digunakan sebagai DLQ. Anda dapat menambahkan konfigurasi DLQ ke setiap wastafel di dalam pipa. Saat pipeline menemukan kesalahan penulisan, ia membuat objek DLQ di bucket S3 yang dikonfigurasi. Objek DLQ ada dalam file JSON sebagai array peristiwa gagal. Pipeline menulis peristiwa ke DLQ ketika salah satu dari kondisi berikut terpenuhi: + Jumlah **percobaan ulang Max** untuk OpenSearch wastafel telah habis. OpenSearch Tertelan membutuhkan minimal 16 untuk pengaturan ini. + Wastafel menolak peristiwa karena kondisi kesalahan. ### Konfigurasi Untuk mengonfigurasi antrian huruf mati untuk sub-pipeline, pilih **Aktifkan DLQ S3** saat Anda mengonfigurasi tujuan wastafel. Kemudian, tentukan pengaturan yang diperlukan untuk antrian. Untuk informasi selengkapnya, lihat [Konfigurasi](https://opensearch.org/docs/latest/data-prepper/pipelines/dlq/#configuration) dalam dokumentasi DLQ Penyedia Data. File yang ditulis ke DLQ S3 ini memiliki pola penamaan berikut: ``` dlq-v${version}-${pipelineName}-${pluginId}-${timestampIso8601}-${uniqueId} ``` Untuk petunjuk mengonfigurasi peran pipeline secara manual agar memungkinkan akses ke bucket S3 yang ditulis oleh DLQ, lihat. [Izin untuk menulis ke Amazon S3 atau antrian surat mati](pipeline-security-overview.md#pipeline-security-dlq) ### Contoh Perhatikan contoh file DLQ berikut: ``` dlq-v2-apache-log-pipeline-opensearch-2023-04-05T15:26:19.152938Z-e7eb675a-f558-4048-8566-dac15a4f8343 ``` Berikut adalah contoh data yang gagal ditulis ke wastafel, dan dikirim ke bucket DLQ S3 untuk analisis lebih lanjut: ``` Record_0 pluginId "opensearch" pluginName "opensearch" pipelineName "apache-log-pipeline" failedData index "logs" indexId null status 0 message "Number of retries reached the limit of max retries (configured value 15)" document log "sample log" timestamp "2023-04-14T10:36:01.070Z" Record_1 pluginId "opensearch" pluginName "opensearch" pipelineName "apache-log-pipeline" failedData index "logs" indexId null status 0 message "Number of retries reached the limit of max retries (configured value 15)" document log "another sample log" timestamp "2023-04-14T10:36:01.071Z" ``` ## Manajemen indeks Amazon OpenSearch Ingestion memiliki banyak kemampuan manajemen indeks, termasuk yang berikut ini. ### Membuat indeks Anda dapat menentukan nama indeks di wastafel pipa dan OpenSearch Ingestion membuat indeks saat menyediakan pipeline. Jika indeks sudah ada, pipeline menggunakannya untuk mengindeks peristiwa yang masuk. Jika Anda menghentikan dan memulai ulang pipeline, atau jika Anda memperbarui konfigurasi YAML-nya, pipeline akan mencoba membuat indeks baru jika belum ada. Pipeline tidak akan pernah bisa menghapus indeks. Contoh berikut sink membuat dua indeks saat pipeline disediakan: ``` sink: - opensearch: index: apache_logs - opensearch: index: nginx_logs ``` ### Menghasilkan nama dan pola indeks Anda dapat menghasilkan nama indeks dinamis dengan menggunakan variabel dari bidang peristiwa yang masuk. Dalam konfigurasi sink, gunakan format `string${}` untuk memberi sinyal interpolasi string, dan gunakan penunjuk JSON untuk mengekstrak bidang dari peristiwa. Pilihan untuk `index_type` adalah `custom` atau`management_disabled`. Karena `index_type` default untuk OpenSearch domain dan `custom` `management_disabled` untuk koleksi OpenSearch Tanpa Server, itu dapat dibiarkan tidak disetel. Misalnya, pipeline berikut memilih `metadataType` bidang dari peristiwa yang masuk untuk menghasilkan nama indeks. ``` pipeline: ... sink: opensearch: index: "metadata-${metadataType}" ``` Konfigurasi berikut terus menghasilkan indeks baru setiap hari atau setiap jam. ``` pipeline: ... sink: opensearch: index: "metadata-${metadataType}-%{yyyy.MM.dd}" pipeline: ... sink: opensearch: index: "metadata-${metadataType}-%{yyyy.MM.dd.HH}" ``` Nama indeks juga bisa berupa string biasa dengan pola tanggal-waktu sebagai akhiran, seperti. `my-index-%{yyyy.MM.dd}` Ketika sink mengirim data ke OpenSearch, itu menggantikan pola tanggal-waktu dengan waktu UTC dan membuat indeks baru untuk setiap hari, seperti. `my-index-2022.01.25` Untuk informasi lebih lanjut, lihat [DateTimeFormatter](https://docs.oracle.com/javase/8/docs/api/java/time/format/DateTimeFormatter.html)kelas. Nama indeks ini juga dapat berupa string yang diformat (dengan atau tanpa akhiran pola tanggal-waktu), seperti. `my-${index}-name` Ketika wastafel mengirim data ke OpenSearch, itu menggantikan `"${index}"` bagian dengan nilai dalam acara yang sedang diproses. Jika formatnya`"${index1/index2/index3}"`, itu menggantikan bidang `index1/index2/index3` dengan nilainya dalam acara tersebut. ### Menghasilkan dokumen IDs Pipeline dapat menghasilkan ID dokumen saat mengindeks dokumen ke OpenSearch. Ini dapat menyimpulkan dokumen ini IDs dari bidang dalam peristiwa yang masuk. Contoh ini menggunakan `uuid` bidang dari peristiwa yang masuk untuk menghasilkan ID dokumen. ``` pipeline: ... sink: opensearch: index_type: custom index: "metadata-${metadataType}-%{yyyy.MM.dd}" "document_id": "uuid" ``` Dalam contoh berikut, prosesor [Tambah entri](https://opensearch.org/docs/latest/data-prepper/pipelines/configuration/processors/add-entries/) menggabungkan bidang `uuid` dan `other_field` dari peristiwa yang masuk untuk menghasilkan ID dokumen. `create`Tindakan ini memastikan bahwa dokumen dengan identik IDs tidak ditimpa. Pipeline menjatuhkan dokumen duplikat tanpa percobaan ulang atau acara DLQ. Ini adalah harapan yang masuk akal bagi penulis pipeline yang menggunakan tindakan ini, karena tujuannya adalah untuk menghindari memperbarui dokumen yang ada. ``` pipeline: ... processor: - add_entries: entries: - key: "my_doc_id_field" format: "${uuid}-${other_field}" sink: - opensearch: ... action: "create" document_id: "my_doc_id" ``` Anda mungkin ingin menyetel ID dokumen acara ke bidang dari sub-objek. Dalam contoh berikut, plugin OpenSearch sink menggunakan sub-objek `info/id` untuk menghasilkan ID dokumen. ``` sink: - opensearch: ... document_id: info/id ``` Mengingat peristiwa berikut, pipeline akan menghasilkan dokumen dengan `_id` bidang yang disetel ke`json001`: ``` { "fieldA":"arbitrary value", "info":{ "id":"json001", "fieldA":"xyz", "fieldB":"def" } } ``` ### Menghasilkan perutean IDs Anda dapat menggunakan `routing_field` opsi dalam plugin OpenSearch sink untuk mengatur nilai properti routing dokumen (`_routing`) ke nilai dari peristiwa yang masuk. Routing mendukung sintaks pointer JSON, sehingga bidang bersarang juga tersedia, bukan hanya bidang tingkat atas. ``` sink: - opensearch: ... routing_field: metadata/id document_id: id ``` Mengingat peristiwa berikut, plugin menghasilkan dokumen dengan `_routing` bidang diatur ke`abcd`: ``` { "id":"123", "metadata":{ "id":"abcd", "fieldA":"valueA" }, "fieldB":"valueB" } ``` Untuk petunjuk membuat templat indeks yang dapat digunakan pipeline selama pembuatan indeks, lihat [Templat indeks](https://opensearch.org/docs/latest/im-plugin/index-templates/). ## End-to-end pengakuan OpenSearch *Penyerapan memastikan daya tahan dan keandalan data dengan melacak pengirimannya dari sumber ke bak cuci di jaringan pipa tanpa kewarganegaraan menggunakan pengakuan. end-to-end* **catatan** Saat ini, hanya plugin [sumber S3](https://opensearch.org/docs/latest/data-prepper/pipelines/configuration/sources/s3/) yang mendukung end-to-end pengakuan. Dengan end-to-end pengakuan, plugin sumber pipeline membuat *set pengakuan untuk memantau sekumpulan* peristiwa. Ini menerima pengakuan positif ketika peristiwa itu berhasil dikirim ke wastafel mereka, atau pengakuan negatif ketika salah satu peristiwa tidak dapat dikirim ke wastafel mereka. Jika terjadi kegagalan atau kerusakan komponen pipa, atau jika sumber gagal menerima pengakuan, sumber akan habis waktu dan mengambil tindakan yang diperlukan seperti mencoba kembali atau mencatat kegagalan. **Jika pipeline memiliki beberapa sink atau beberapa sub-pipeline yang dikonfigurasi, pengakuan tingkat peristiwa dikirim hanya setelah acara dikirim ke semua sink di semua sub-pipeline.** Jika wastafel memiliki DLQ yang dikonfigurasi, end-to-end pengakuan juga melacak peristiwa yang ditulis ke DLQ. Untuk mengaktifkan end-to-end pengakuan, perluas **Opsi tambahan** di konfigurasi sumber Amazon S3 dan **pilih end-to-end** Aktifkan pengakuan pesan. ## Sumber tekanan balik Pipa dapat mengalami tekanan balik saat sibuk memproses data, atau jika sink sementara turun atau lambat untuk menelan data. OpenSearch Ingestion memiliki cara yang berbeda untuk menangani tekanan balik tergantung pada plugin sumber yang digunakan pipa. ### Sumber HTTP Saluran pipa yang menggunakan plugin [sumber HTTP](https://opensearch.org/docs/latest/data-prepper/pipelines/configuration/sources/http-source/) menangani tekanan balik secara berbeda tergantung pada komponen pipa mana yang padat: + **Buffer** — Ketika buffer penuh, pipeline mulai mengembalikan status HTTP `REQUEST_TIMEOUT` dengan kode kesalahan 408 kembali ke titik akhir sumber. Saat buffer dibebaskan, pipeline mulai memproses peristiwa HTTP lagi. + **Thread sumber** — Ketika semua thread sumber HTTP sibuk mengeksekusi permintaan dan ukuran antrian permintaan yang belum diproses telah melebihi jumlah maksimum permintaan yang diizinkan, pipeline mulai mengembalikan status HTTP `TOO_MANY_REQUESTS` dengan kode kesalahan 429 kembali ke titik akhir sumber. Ketika antrian permintaan turun di bawah ukuran antrian maksimum yang diizinkan, pipeline mulai memproses permintaan lagi. ### OTel sumber Ketika buffer penuh untuk pipeline yang menggunakan OpenTelemetry sumber ([OTel log](https://github.com/opensearch-project/data-prepper/tree/main/data-prepper-plugins/otel-logs-source), [OTel metrik](https://opensearch.org/docs/latest/data-prepper/pipelines/configuration/sources/otel-metrics-source/), dan [OTel jejak](https://opensearch.org/docs/latest/data-prepper/pipelines/configuration/sources/otel-trace/)), pipeline mulai mengembalikan status HTTP `REQUEST_TIMEOUT` dengan kode kesalahan 408 ke titik akhir sumber. Saat buffer dibebaskan, pipeline mulai memproses peristiwa lagi. ### Sumber S3 Ketika buffer penuh untuk saluran pipa dengan sumber [S3](https://opensearch.org/docs/latest/data-prepper/pipelines/configuration/sources/s3/), saluran pipa berhenti memproses pemberitahuan SQS. Saat buffer dibebaskan, saluran pipa mulai memproses notifikasi lagi. Jika sink mati atau tidak dapat menyerap data dan end-to-end pengakuan diaktifkan untuk sumber, pipeline berhenti memproses notifikasi SQS hingga menerima pengakuan yang berhasil dari semua sink.