从 Amazon S3 表中加载流数据

创建目录。在此示例中，我们使用名称 s3-to-opensearch。

在名为 sample.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)

编辑 region 和 host 的变量。

安装 pip（如果您尚未安装，则将依赖项安装到 package 目录：

cd s3-to-opensearch

pip install --target ./package requests
pip install --target ./package requests_aws4auth

所有 Lambda 执行环境都已安装 Boto3，因此无需将其包含在部署程序包中。

打包应用程序代码和依赖项：

cd package
zip -r ../lambda.zip .

cd ..
zip -g lambda.zip sample.py

选择添加触发器并选择 S3。

选择存储桶。

对于 Event type (事件类型)，选择 PUT。

对于 Prefix (前缀)，键入 logs/。

对于后缀，键入 .log。

确认递归调用警告，然后选择添加。

选择上载自和 .zip 文件，然后按照提示上传部署程序包。

上载完成后，编辑 Runtime 设置并更改处理程序为 sample.handler。此设置告知 Lambda 在触发之后应执行的文件 (sample.py) 和方法 (handler)。