捕获 AWS lambda 函数的输出并通过 SNS 通过电子邮件发送

Question

Lambda 函数代码：

import json
import datetime
import redshift_module.pygresql_redshift_common as db_handler
from decouple import config
import pg

host = config('host')
port = config('port')
db_name = config('db_name')
db_user = config('db_user')
db_password = config('db_password')


def get_connection(host, port, dbname, user, password):
    rs_conn_string = "host=%s port=%s dbname=%s user=%s password=%s" % (
        host, port, dbname, user, password)
    rs_conn = pg.connect(dbname=rs_conn_string)
    rs_conn.query("set statement_timeout = 1200000")
    return rs_conn


def query(con, query):
    res = con.query(query)
    return res


rs_conn = db_handler.get_connection(host, port, db_name, db_user, db_password)
query_string = "call XYZ_warehouse.ABC_revenue_last_7_days();"

res = query(rs_conn, query_string)
print(res.getresults())


def lambda_handler(event, context):
    return {'statusCode': 200, 'body': json.dumps('Hello from Lambda!')}

代码在 redshift 数据库上运行查询函数。 lambda 的输出包含如下字符串：

DEUS revenue is completely matching and revenue is : 5139 and Revenue without shipping is: 4987

我需要做的是捕获执行结果中的此类行，也许将它们存储在变量中或日志文件中的某处，以便我可以通过电子邮件发送它们。我尝试在查询对象上运行 .getresults() 但它返回一个空字符串。也许是因为它没有返回记录或表格？

有没有办法捕获 lambda 函数的完整输出？或者将其发送到某处的文件中？

Answer 1

创建字典并保存您想要通过电子邮件发送的所有输出。

现在将其转换为 json 并通过 SNS 或 SES 发送

Answer 2

您的问题有点不清楚，您是否还需要检索查询结果方面的帮助？

如果是这样，我建议您从此处阅读

有没有办法捕获 lambda 函数的完整输出？或者将其发送到某个文件？

一种选择是将 lambda func 的输出发送到 s3 存储桶，例如

import boto3

s3 = boto3.resource(
    's3',
    region_name='us-east-1',
    aws_access_key_id=KEY_ID,
    aws_secret_access_key=ACCESS_KEY
)
content="String content to write to a new S3 file"
s3.Object('my-bucket-name', 'newfile.txt').put(Body=content)

您可以设置一个 s3 触发器，然后 将新对象添加到存储桶后立即触发第二个 lambda。

然后，您的第二个 lambda 可以转换为 JSON 并通过 SNS/SES 作为 Vaquar Khan 的答案发送。

Answer 3

如果我理解正确的话“你想将 lambda 的输出存储在单独的位置/文件中”以将其用作电子邮件。

如果它是纯 lambda 输出，您可以简单地使用 lambda 目标。

目标允许您根据失败或成功来分离结果。如果成功，您可以将其发送到目标（SNS、SQS、EventBridge 或不同的 Lambda），如果失败，您可以将其发送到失败目的地

目标文档

Answer 4

创建

作为初学者，我想到的解决方案是：

• 另一个由 CloudwatchLogs 触发的 lambda 函数。
• 选择输出查询结果的函数的日志组。它的日志将包含您需要的行。
• 要从日志中获取这些行，您需要以下代码：

  • decoded_data = json.loads(gzip.decompress(b64decode(event['awslogs']['data'])))
    

  • 从查询 lambda 函数创建的最新日志将存储在 decoded_data
  • 然后从词典中提取所需的行并通过 SNS 通过电子邮件发送。