AWS Lambda¶

借助 AWS Lambda，您无需配置或管理服务器即可运行代码。您只需为使用的计算时间付费，当您的代码未运行时不收费。您可以运行几乎任何类型的应用程序或后端服务的代码，所有这些都无需管理。只需上传您的代码，Lambda 就会处理运行和扩展代码所需的一切，并确保高可用性。您可以将代码设置为自动从其他 AWS 服务触发，或从任何 Web 或移动应用程序直接调用它。

前提任务¶

要使用这些操作符，您必须执行以下操作

使用 AWS 控制台或 AWS CLI 创建必要的资源。
通过 pip 安装 API 库。
pip install 'apache-airflow[amazon]'
详细信息请参阅 Airflow™ 安装
设置连接.

通用参数¶

aws_conn_id

对 Amazon Web Services 连接 ID 的引用。如果此参数设置为 None，则使用默认的 boto3 行为，不进行连接查找。否则，使用存储在连接中的凭据。默认值：aws_default

region_name

AWS 区域名称。如果此参数设置为 None 或省略，则将使用 AWS 连接额外参数中的 region_name。否则，使用指定的值而不是连接值。默认值：None

verify

是否验证 SSL 证书。

False - 不验证 SSL 证书。
path/to/cert/bundle.pem - 要使用的 CA 证书包的文件名。如果您想使用与 botocore 使用的不同 CA 证书包，则可以指定此参数。

如果此参数设置为 None 或省略，则将使用 AWS 连接额外参数中的 verify。否则，使用指定的值而不是连接值。默认值：None

botocore_config

提供的字典用于构造 botocore.config.Config。此配置可用于配置避免限制异常、超时等。

示例，有关参数的更多详细信息，请查看 botocore.config.Config¶

{
    "signature_version": "unsigned",
    "s3": {
        "us_east_1_regional_endpoint": True,
    },
    "retries": {
      "mode": "standard",
      "max_attempts": 10,
    },
    "connect_timeout": 300,
    "read_timeout": 300,
    "tcp_keepalive": True,
}

如果此参数设置为 None 或省略，则将使用 AWS 连接额外参数中的 config_kwargs。否则，使用指定的值而不是连接值。默认值：None

注意

指定空字典 {} 将覆盖 botocore.config.Config 的连接配置

操作符¶

创建 AWS Lambda 函数¶

要创建 AWS lambda 函数，您可以使用 LambdaCreateFunctionOperator。可以通过传递参数 deferrable=True 以可延迟模式运行此操作符。这需要安装 aiobotocore 模块。

tests/system/providers/amazon/aws/example_lambda.py[源代码]

create_lambda_function = LambdaCreateFunctionOperator(
    task_id="create_lambda_function",
    function_name=lambda_function_name,
    runtime="python3.9",
    role=role_arn,
    handler="lambda_function.test",
    code={
        "ZipFile": create_zip(CODE_CONTENT),
    },
)

调用 AWS Lambda 函数¶

要调用 AWS lambda 函数，您可以使用 LambdaInvokeFunctionOperator。

注意

根据 Lambda.Client.invoke 文档，对于具有较长超时的同步调用（invocation_type="RequestResponse"），您的客户端在等待响应时可能会在同步调用期间断开连接。

如果发生这种情况，您将看到类似于以下内容的 ReadTimeoutError 异常

urllib3.exceptions.ReadTimeoutError: AWSHTTPSConnectionPool(host='lambda.us-east-1.amazonaws.com', port=443): Read timed out. (read timeout=60)

如果遇到此问题，则需要提供 botocore.config.Config 以使用具有超时或保持活动设置的长连接。

通过提供 botocore_config 作为操作符参数

{
  "connect_timeout": 900,
  "read_timeout": 900,
  "tcp_keepalive": True,
}

或在关联的 AWS 连接额外参数中指定 config_kwargs

{
  "config_kwargs": {
    "connect_timeout": 900,
    "read_timeout": 900,
    "tcp_keepalive": true
  }
}

此外，您可能需要配置防火墙、代理或操作系统以允许具有超时或保持活动设置的长连接。

另请参阅

注意

您无法描述 AWS Lambda 函数的异步调用（invocation_type="Event"）。唯一的方法是为异步调用配置目标并感知目标。

tests/system/providers/amazon/aws/example_lambda.py[源代码]

invoke_lambda_function = LambdaInvokeFunctionOperator(
    task_id="invoke_lambda_function",
    function_name=lambda_function_name,
    payload=json.dumps({"SampleEvent": {"SampleData": {"Name": "XYZ", "DoB": "1993-01-01"}}}),
)

传感器¶

等待 AWS Lambda 函数部署状态¶

要检查 AWS Lambda 函数的部署状态，直到它达到目标状态或另一个终端状态，您可以使用 LambdaFunctionStateSensor。

tests/system/providers/amazon/aws/example_lambda.py[源代码]

wait_lambda_function_state = LambdaFunctionStateSensor(
    task_id="wait_lambda_function_state",
    function_name=lambda_function_name,
)

参考¶

Lambda 的 AWS boto3 库文档