从 Airflow 触发 AWS Lambda 函数

Question

我在 AWS lambda 中创建了一个函数，如下所示：

import boto3
import numpy as np
import pandas as pd
import s3fs
from io import StringIO


def test(event=None, context=None):
    # creating a pandas dataframe from an api
    # placing 2 csv files in S3 bucket

此函数查询外部 API 并将 2 个csv文件放入 S3 存储桶中。我想在Airflow中触发这个功能，我找到了这段代码：

import boto3, json, typing

def invokeLambdaFunction(*, functionName:str=None, payload:typing.Mapping[str, str]=None):
    if  functionName == None:
        raise Exception('ERROR: functionName parameter cannot be NULL')
    payloadStr = json.dumps(payload)
    payloadBytesArr = bytes(payloadStr, encoding='utf8')
    client = boto3.client('lambda')
    response = client.invoke(
        FunctionName=test,
        InvocationType="RequestResponse",
        Payload=payloadBytesArr
    )
    return response



if __name__ == '__main__':
    payloadObj = {"something" : "1111111-222222-333333-bba8-1111111"}
    response = invokeLambdaFunction(functionName='test',  payload=payloadObj)
    print(f'response:{response}')

但据我了解，此代码片段未连接到 S3。这是从 Airflow 触发 AWS Lambda 函数的正确方法还是有更好的方法？

Answer 1

我建议您使用AwsLambdaHook：

https://airflow.apache.org/docs/stable/_api/airflow/contrib/hooks/aws_lambda_hook/index.html#module-airflow.contrib.hooks.aws_lambda_hook

有一个测试显示了它在触发 Lambda 函数中的用途：

https://github.com/apache/airflow/blob/master/tests/providers/amazon/aws/hooks/test_lambda_function.py