View a markdown version of this page

Menggunakan pipeline OpenSearch Ingestion dengan Amazon S3 - OpenSearch Layanan Amazon
Amazon S3 sebagai sumberAmazon S3 sebagai tujuanAkun silang Amazon S3 sebagai sumber

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

Menggunakan pipeline OpenSearch Ingestion dengan Amazon S3

Dengan OpenSearch Ingestion, Anda dapat menggunakan Amazon S3 sebagai sumber atau sebagai tujuan. Saat menggunakan Amazon S3 sebagai sumber, Anda mengirim data ke pipeline Ing OpenSearch estion. Saat menggunakan Amazon S3 sebagai tujuan, Anda menulis data dari pipeline Ing OpenSearch estion ke satu atau beberapa bucket S3.

Amazon S3 sebagai sumber

Ada dua cara yang dapat Anda gunakan Amazon S3 sebagai sumber untuk memproses data—dengan S3-SQS pemrosesan dan pemindaian ter jadwal.

Gunakan S3-SQS pemrosesan saat Anda memerlukan pemindaian hampir real-time file setelah ditulis ke S3. Anda dapat mengonfigurasi bucket Amazon S3 untuk memunculkan peristiwa setiap kali objek disimpan atau dimodifikasi di dalam bucket. Gunakan pemindaian terjadwal satu kali atau berulang untuk memproses data batch dalam bucket S3.

Prasyarat

Untuk menggunakan Amazon S3 sebagai sumber saluran Inges OpenSearch tion untuk pemindaian atau S3-SQS pemrosesan terjadwal, pertama-tama buat bucket S3.

catatan

Jika bucket S3 yang digunakan sebagai sumber di OpenSearch pipeline Ingestion berbeda Akun AWS, Anda juga perlu mengaktifkan izin baca lintas akun di bucket. Hal ini memungkinkan pipeline untuk membaca dan memproses data. Untuk mengaktifkan izin lintas akun, lihat Pemilik Bucket memberikan izin bucket lintas-akun di Panduan Pengguna Amazon S3.

Jika bucket S3 Anda ada di beberapa akun, gunakan bucket_owners peta. Sebagai contoh, lihat akses Cross-account S3 dalam OpenSearch dokumentasi.

Untuk mengatur S3-SQS pemrosesan, Anda juga perlu melakukan langkah-langkah berikut:

  1. Buat antri an Amazon SQS.

  2. Aktifkan notifikasi peristiwa pada bucket S3 dengan antrian SQS sebagai tujuan.

Langkah 1: Konfigurasikan peran pipeline

Tidak seperti plugin sumber lain yang mendorong data ke pipeline, plugin sumber S3 memiliki arsitektur berbasis baca di mana pipeline menarik data dari sumbernya.

Oleh karena itu, agar pipeline dapat dibaca dari S3, Anda harus menentukan peran dalam konfigurasi sumber S3 pipeline yang memiliki akses ke bucket S3 dan antrian Amazon SQS. Pipeline akan mengambil peran ini untuk membaca data dari antrian.

catatan

Peran yang Anda tentukan dalam konfigurasi sumber S3 haruslah peran pipeline. Oleh karena itu, peran pipeline Anda harus berisi dua kebijakan izin terpisah—satu untuk menulis ke wastafel, dan satu lagi untuk menarik dari sumber S3. Anda harus menggunakan yang sama sts_role_arn di semua komponen pipa.

Kebijakan sampel berikut menunjukkan izin yang diperlukan untuk menggunakan S3 sebagai sumber:

JSON
{
  "Version":"2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action":[
          "s3:ListBucket",
          "s3:GetBucketLocation",
          "s3:GetObject"
       ],
      "Resource": "arn:aws:s3:::amzn-s3-demo-bucket/*"
    },
    {
       "Effect":"Allow",
       "Action":"s3:ListAllMyBuckets",
       "Resource":"arn:aws:s3:::*"
    },
    {
      "Effect": "Allow",
      "Action": [
        "sqs:DeleteMessage",
        "sqs:ReceiveMessage",
        "sqs:ChangeMessageVisibility"
      ],
      "Resource": "arn:aws:sqs:us-east-1:111122223333:MyS3EventSqsQueue"
    }
  ]
}

Anda harus melampirkan izin ini ke peran IAM yang Anda tentukan dalam sts_role_arn opsi dalam konfigurasi plugin sumber S3:

version: "2"
source:
  s3:
    ...
    aws:
      ...
processor:
  ...
sink:
  - opensearch:
      ...

Langkah 2: Buat pipa

Setelah menyiapkan izin, Anda dapat mengonfigurasi pipeline Ing OpenSearch estion tergantung pada kasus penggunaan Amazon S3 Anda.

S3-SQS pengolahan

Untuk mengatur S3-SQS pemrosesan, konfigurasikan pipeline Anda untuk menentukan S3 sebagai sumber dan atur notifikasi Amazon SQS:

version: "2"
s3-pipeline:
  source:
    s3:
      notification_type: "sqs"
      codec:
        newline: null
      sqs:
        queue_url: "https://sqs.us-east-1amazonaws.com/account-id/ingestion-queue"
      compression: "none"
      aws:
        region: "region"
  processor:
  - grok:
      match:
        message:
        - "%{COMMONAPACHELOG}"
  - date:
      destination: "@timestamp"
      from_time_received: true
  sink:
  - opensearch:
      hosts: ["https://search-domain-endpoint.us-east-1es.amazonaws.com"]
      index: "index-name"
      aws:
        region: "region"

Jika Anda mengamati pemanfaatan CPU yang rendah saat memproses file kecil di Amazon S3, pertimbangkan untuk meningkatkan throughput dengan memodifikasi nilai workers opsi. Untuk informasi selengkapnya, lihat opsi konfigurasi plugin S3.

Pemindaian terjadwal

Untuk mengatur pemindaian terjadwal, konfigurasikan pipeline Anda dengan jadwal di tingkat pemindaian yang berlaku untuk semua bucket S3 Anda, atau pada tingkat bucket. Jadwal tingkat bucket atau konfigurasi interval pemindaian selalu mengganti konfigurasi tingkat pemindaian.

Anda dapat mengonfigurasi pemindaian terjadwal dengan pemindaian satu kali, yang ideal untuk migrasi data, atau pemindaian berulang, yang ideal untuk pemrosesan batch.

Untuk mengonfigurasi pipeline Anda agar dibaca dari Amazon S3, gunakan cetak biru Amazon S3 yang telah dikonfigurasi sebelumnya. Anda dapat mengedit scan bagian konfigurasi pipeline untuk memenuhi kebutuhan penjadwalan Anda. Untuk informasi selengkapnya, lihat Bekerja dengan cetak biru.

One-time memindai

Pemindaian terjadwal satu kali berjalan sekali. Dalam konfigurasi pipeline Anda, Anda dapat menggunakan start_time a dan end_time untuk menentukan kapan Anda ingin objek dalam bucket dipindai. Atau, Anda dapat menggunakan range untuk menentukan interval waktu relatif terhadap waktu saat ini yang ingin objek dalam bucket dipindai.

Misalnya, rentang yang diatur untuk mem PT4H indai semua file yang dibuat dalam empat jam terakhir. Untuk mengonfigurasi pemindaian satu kali untuk menjalankan kedua kalinya, Anda harus menghentikan dan memulai ulang pipeline. Jika Anda tidak memiliki rentang yang dikonfigurasi, Anda juga harus memperbarui waktu mulai dan akhir.

Konfigurasi berikut menyiapkan pemindaian satu kali untuk semua bucket dan semua objek dalam bucket tersebut:

version: "2"
log-pipeline:
  source:
    s3:
      codec:
        csv:
      compression: "none"
      aws:
        region: "region"
      acknowledgments: true
      scan:
        buckets:
          - bucket:
              name: my-bucket
              filter:
                include_prefix:
                  - Objects1/
                exclude_suffix:
                  - .jpeg
                  - .png
          - bucket:
              name: my-bucket-2
              key_prefix:
                include:
                  - Objects2/
                exclude_suffix:
                  - .jpeg
                  - .png
      delete_s3_objects_on_read: false
  processor:
    - date:
        destination: "@timestamp"
        from_time_received: true
  sink:
    - opensearch:
        hosts: ["https://search-domain-endpoint.us-east-1es.amazonaws.com"]
        index: "index-name"
        aws:
          region: "region"
        dlq:
          s3:
            bucket: "dlq-bucket"
            region: "us-east-1"

Konfigurasi berikut menyiapkan pemindaian satu kali untuk semua bucket selama jendela waktu tertentu. Ini berarti bahwa S3 hanya memproses objek-objek dengan waktu pembuatan yang berada dalam jendela ini.

scan:
  start_time: 2023-01-21T18:00:00.000Z
  end_time: 2023-04-21T18:00:00.000Z
  buckets:
    - bucket:
        name: my-bucket-1
        filter:
          include:
            - Objects1/
          exclude_suffix:
            - .jpeg
            - .png
    - bucket:
        name: my-bucket-2
        filter:
          include:
            - Objects2/
          exclude_suffix:
            - .jpeg
            - .png

Konfigurasi berikut mengatur pemindaian satu kali pada tingkat pemindaian dan tingkat bucket. Waktu mulai dan berakhir di tingkat bucket mengesampingkan waktu mulai dan berakhir di tingkat pemindaian. 

scan:
  start_time: 2023-01-21T18:00:00.000Z
  end_time: 2023-04-21T18:00:00.000Z
  buckets:
    - bucket:
        start_time: 2023-01-21T18:00:00.000Z
        end_time: 2023-04-21T18:00:00.000Z
        name: my-bucket-1
        filter:
          include:
            - Objects1/
          exclude_suffix:
            - .jpeg
            - .png
    - bucket:
        start_time: 2023-01-21T18:00:00.000Z
        end_time: 2023-04-21T18:00:00.000Z
        name: my-bucket-2
        filter:
          include:
            - Objects2/
          exclude_suffix:
            - .jpeg
            - .png

Menghentikan pipa menghapus referensi yang sudah ada sebelumnya dari objek apa yang telah dipindai oleh pipa sebelum berhenti. Jika pipa pemindaian tunggal dihentikan, itu akan memindai ulang semua objek lagi setelah dimulai, bahkan jika mereka sudah dipindai. Jika Anda perlu menghentikan pipa pemindaian tunggal, disarankan Anda mengubah jendela waktu Anda sebelum memulai pipa lagi.

Jika Anda perlu memfilter objek berdasarkan waktu mulai dan waktu akhir, menghentikan dan memulai pipeline Anda adalah satu-satunya pilihan. Jika Anda tidak perlu memfilter berdasarkan waktu mulai dan waktu akhir, Anda dapat memfilter objek berdasarkan nama. Berkeliaran dengan nama tidak mengharuskan Anda untuk berhenti dan memulai pipeline Anda. Untuk melakukan ini, gunakan include_prefix danexclude_suffix.

Pemindaian berulang

Pemindaian terjadwal berulang menjalankan pemindaian bucket S3 yang Anda tentukan secara berkala dan terjadwal. Anda hanya dapat mengonfigurasi interval ini pada tingkat pemindaian karena konfigurasi tingkat bucket individual tidak didukung.

Dalam konfigurasi pipeline Anda, interval menentukan frekuensi pemindaian berulang, dan bisa antara 30 detik dan 365 hari. Pemindaian pertama ini selalu terjadi saat Anda membuat pipeline. Men count definisikan jumlah total instance pemindaian.

Konfigurasi berikut mengatur pemindaian berulang, dengan penundaan 12 jam antara pemindaian:

scan:
  scheduling:
    interval: PT12H
    count: 4
  buckets:
    - bucket:
        name: my-bucket-1
        filter:
          include:
            - Objects1/
          exclude_suffix:
            - .jpeg
            - .png
    - bucket:
        name: my-bucket-2
        filter:
          include:
            - Objects2/
          exclude_suffix:
            - .jpeg
            - .png

Amazon S3 sebagai tujuan

Untuk menulis data dari pipeline OpenSearch Ingestion ke bucket S3, gunakan cetak biru S3 yang telah dikonfigurasi sebelumnya untuk membuat pipeline dengan sink S3. Pipa ini merutekan data selektif ke OpenSearch wastafel dan secara bersamaan mengirimkan semua data untuk arsip di S3. Untuk informasi selengkapnya, lihat Bekerja dengan cetak biru.

Saat Anda membuat wastafel S3, Anda dapat menentukan pemformatan pilihan Anda dari berbagai codec sink. Misalnya, jika Anda ingin menulis data dalam format kolom, pilih codec Parquet atau Avro. Jika Anda lebih suka format berbasis baris, pilih JSON atau NDJSON. Untuk menulis data ke S3 dalam skema tertentu, Anda juga dapat menentukan skema inline dalam sink codec menggunakan format Avro.

Contoh berikut mendefinisikan skema sebaris di wastafel S3:

- s3:
  codec:
    parquet:
      schema: >
        {
           "type" : "record",
           "namespace" : "org.vpcFlowLog.examples",
           "name" : "VpcFlowLog",
           "fields" : [
             { "name" : "version", "type" : "string"},
             { "name" : "srcport", "type": "int"},
             { "name" : "dstport", "type": "int"},
             { "name" : "start", "type": "int"},
             { "name" : "end", "type": "int"},
             { "name" : "protocol", "type": "int"},
             { "name" : "packets", "type": "int"},
             { "name" : "bytes", "type": "int"},
             { "name" : "action", "type": "string"},
             { "name" : "logStatus", "type" : "string"}
           ]
         }

Saat Anda mendefinisikan skema ini, tentukan superset dari semua kunci yang mungkin ada dalam berbagai jenis peristiwa yang dikirimkan pipeline Anda ke wastafel.

Misalnya, jika suatu peristiwa memiliki kemungkinan kunci hilang, tambahkan kunci itu dalam skema Anda dengan null nilai. Deklarasi nilai nol memungkinkan skema untuk memproses data yang tidak seragam (di mana beberapa peristiwa memiliki kunci ini dan yang lain tidak). Ketika peristiwa yang masuk memang memiliki kunci-kunci ini, nilainya ditulis ke sink.

Definisi skema ini bertindak sebagai filter yang hanya memungkinkan kunci yang ditentukan dikirim ke sink, dan melepaskan kunci yang tidak ditentukan dari peristiwa yang masuk.

Anda juga dapat menggunakan include_keys dan exclude_keys di wastafel Anda untuk menyaring data yang dialihkan ke wastafel lain. Kedua filter ini saling eksklusif, jadi Anda hanya dapat menggunakan satu per satu dalam skema Anda. Selain itu, Anda tidak dapat menggunakannya dalam skema yang ditentukan pengguna.

Untuk membuat saluran pipa dengan filter seperti itu, gunakan cetak biru filter wastafel yang telah dikonfigurasi sebelumnya. Untuk informasi selengkapnya, lihat Bekerja dengan cetak biru.

Akun silang Amazon S3 sebagai sumber

Anda dapat memberikan akses di seluruh akun dengan Amazon S3 sehingga OpenSearch pipeline Ingestion dapat mengakses bucket S3 di akun lain sebagai sumber. Untuk mengaktifkan akses lintas akun, lihat Pemilik Bucket memberikan izin bucket lintas-akun di Panduan Pengguna Amazon S3. Setelah Anda memberikan akses, pastikan peran pipeline Anda memiliki izin yang diperlukan.

Kemudian, Anda dapat membuat pipeline menggunakan bucket_owners untuk mengaktifkan akses lintas akun ke bucket Amazon S3 sebagai sumber:

s3-pipeline:
 source:
  s3:
   notification_type: "sqs"
   codec:
    csv:
     delimiter: ","
     quote_character: "\""
     detect_header: True
   sqs:
    queue_url: "https://sqs.ap-northeast-1.amazonaws.com/401447383613/test-s3-queue"
   bucket_owners:
    my-bucket-01: 123456789012
    my-bucket-02: 999999999999
   compression: "gzip"