Memuat data streaming dari Amazon S3

Anda dapat menggunakan Lambda untuk mengirim data ke domain OpenSearch Layanan Anda dari Amazon S3. Data baru yang tiba di bucket S3 memicu notifikasi peristiwa untuk Lambda, yang kemudian menjalankan kode kustom Anda untuk melakukan pengindeksan.

Metode data streaming ini sangat fleksibel. Anda dapat mengindeks metadata objek, atau jika objek plaintext, mengurai dan mengindeks beberapa elemen dari tubuh objek. Bagian ini mencakup beberapa kode sampel Python yang kurang modern yang menggunakan ekspresi reguler untuk mengurai file log dan mengindeks kecocokan.

Prasyarat

Sebelum melanjutkan, Anda harus memiliki sumber daya berikut.

Prasyarat	Deskripsi
Bucket Amazon S3	Untuk informasi selengkapnya, lihat Membuat bucket S3 pertama Anda di Panduan Pengguna Layanan Penyimpanan Sederhana Amazon. Bucket harus berada di Wilayah yang sama dengan domain OpenSearch Layanan Anda.
OpenSearch Domain layanan	Tujuan untuk data setelah fungsi Lambda Anda memprosesnya. Untuk informasi selengkapnya, lihat Membuat domain OpenSearch Layanan.

Membuat paket deployment Lambda

Paket deployment adalah file ZIP atau JAR yang berisi kode Anda dan dependensinya. Bagian ini mencakup kode sampel Python. Untuk bahasa pemrograman lainnya, lihat Paket penyebaran Lambda di Panduan Pengembang.AWS Lambda

1. Buatlah sebuah direktori. Dalam sampel ini, kita menggunakan nama s3-to-opensearch.

2. Buat file dalam direktori bernamasample.py:

   import boto3
   import re
   import requests
   from requests_aws4auth import AWS4Auth
   
   region = '' # e.g. us-west-1
   service = 'es'
   credentials = boto3.Session().get_credentials()
   awsauth = AWS4Auth(credentials.access_key, credentials.secret_key, region, service, session_token=credentials.token)
   
   host = '' # the OpenSearch Service domain, e.g. https://search-mydomain.us-west-1.es.amazonaws.com
   index = 'lambda-s3-index'
   datatype = '_doc'
   url = host + '/' + index + '/' + datatype
   
   headers = { "Content-Type": "application/json" }
   
   s3 = boto3.client('s3')
   
   # Regular expressions used to parse some simple log lines
   ip_pattern = re.compile('(\d+\.\d+\.\d+\.\d+)')
   time_pattern = re.compile('\[(\d+\/\w\w\w\/\d\d\d\d:\d\d:\d\d:\d\d\s-\d\d\d\d)\]')
   message_pattern = re.compile('\"(.+)\"')
   
   # Lambda execution starts here
   def handler(event, context):
       for record in event['Records']:
   
           # Get the bucket name and key for the new file
           bucket = record['s3']['bucket']['name']
           key = record['s3']['object']['key']
   
           # Get, read, and split the file into lines
           obj = s3.get_object(Bucket=bucket, Key=key)
           body = obj['Body'].read()
           lines = body.splitlines()
   
           # Match the regular expressions to each line and index the JSON
           for line in lines:
               line = line.decode("utf-8")
               ip = ip_pattern.search(line).group(1)
               timestamp = time_pattern.search(line).group(1)
               message = message_pattern.search(line).group(1)
   
               document = { "ip": ip, "timestamp": timestamp, "message": message }
               r = requests.post(url, auth=awsauth, json=document, headers=headers)

   Edit variabel untuk region dan host.

3. Instal pip jika Anda belum melakukannya, lalu instal dependensi ke direktori baru: package

   cd s3-to-opensearch
   
   pip install --target ./package requests
   pip install --target ./package requests_aws4auth

   Semua lingkungan eksekusi Lambda telah menginstal Boto3, sehingga Anda tidak perlu memasukkannya ke dalam paket deployment.

4. Paket kode aplikasi dan dependensi:

   cd package
   zip -r ../lambda.zip .
   
   cd ..
   zip -g lambda.zip sample.py

Buat fungsi Lambda

Setelah Anda membuat paket deployment, Anda dapat membuat fungsi Lambda. Saat Anda membuat fungsi, pilih nama, runtime (misalnya, Python 3.8), dan peran IAM. IAM role mendefinisikan izin untuk fungsi Anda. Untuk petunjuk mendetail, lihat Membuat fungsi Lambda dengan konsol di Panduan AWS Lambda Pengembang.

Contoh ini mengasumsikan Anda menggunakan konsol. Pilih Python 3.9 dan peran yang memiliki izin baca S3 dan izin menulis OpenSearch Layanan, seperti yang ditunjukkan pada gambar berikut:

Setelah Anda membuat fungsi, Anda harus menambahkan pemicu. Untuk contoh ini, kita ingin kode berjalan setiap kali file log tiba di bucket S3:

1. Pilih Tambah pemicu dan pilih S3.

2. Pilih bucket Anda.

3. Untuk Jenis peristiwa, pilih TEMPATKAN.

4. Untuk Prefiks, ketik logs/.

5. Untuk Sufiks, ketik.log.

6. Akui peringatan pemanggilan rekursif dan pilih Tambah.

Akhirnya, Anda dapat mengunggah paket deployment Anda:

1. Pilih Unggah dari dan file.zip, lalu ikuti petunjuk untuk mengunggah paket penerapan Anda.

2. Setelah unggahan selesai, edit pengaturan Runtime dan ubah Handler menjadi. sample.handler Pengaturan ini memberitahu Lambda file (sample.py) dan metode (handler) yang harus dijalankan setelah pemicu aktif.

Pada titik ini, Anda memiliki satu set lengkap sumber daya: bucket untuk file log, fungsi yang berjalan setiap kali file log ditambahkan ke bucket, kode yang melakukan penguraian dan pengindeksan, dan domain OpenSearch Layanan untuk pencarian dan visualisasi.

Menguji fungsi Lambda

Setelah Anda membuat fungsi, Anda dapat mengujinya dengan mengunggah file ke bucket Amazon S3. Buat file bernama sample.log dengan menggunakan baris log sampel berikut:

12.345.678.90 - [10/Oct/2000:13:55:36 -0700] "PUT /some-file.jpg"
12.345.678.91 - [10/Oct/2000:14:56:14 -0700] "GET /some-file.jpg"

Unggah file ke folder logs dari bucket S3 Anda. Untuk petunjuk, lihat Mengunggah objek ke bucket Anda di Panduan Pengguna Layanan Penyimpanan Sederhana Amazon.

Kemudian gunakan konsol OpenSearch Layanan atau OpenSearch Dasbor untuk memverifikasi bahwa lambda-s3-index indeks berisi dua dokumen. Anda juga dapat membuat permintaan pencarian standar:

GET https://domain-name/lambda-s3-index/_search?pretty
{
  "hits" : {
    "total" : 2,
    "max_score" : 1.0,
    "hits" : [
      {
        "_index" : "lambda-s3-index",
        "_type" : "_doc",
        "_id" : "vTYXaWIBJWV_TTkEuSDg",
        "_score" : 1.0,
        "_source" : {
          "ip" : "12.345.678.91",
          "message" : "GET /some-file.jpg",
          "timestamp" : "10/Oct/2000:14:56:14 -0700"
        }
      },
      {
        "_index" : "lambda-s3-index",
        "_type" : "_doc",
        "_id" : "vjYmaWIBJWV_TTkEuCAB",
        "_score" : 1.0,
        "_source" : {
          "ip" : "12.345.678.90",
          "message" : "PUT /some-file.jpg",
          "timestamp" : "10/Oct/2000:13:55:36 -0700"
        }
      }
    ]
  }
}