Amazon Athena SQL¶

Amazon Athena 是一种交互式查询服务，可让您使用标准 SQL 轻松分析 Amazon Simple Storage Service (S3) 中的数据。 Athena 是无服务器的，因此无需设置或管理任何基础设施，您只需为运行的查询付费。要开始使用，只需指向 S3 中的数据，定义模式，然后使用标准 SQL 开始查询即可。

先决条件任务¶

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

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

操作符¶

执行 SQL 查询¶

通用的 SQLExecuteQueryOperator 可用于使用 Athena 连接针对 Amazon Athena 执行 SQL 查询。

要在不将结果返回到 Airflow 的情况下针对 Amazon Athena 执行单个 SQL 查询，请改用 AthenaOperator。

tests/system/common/sql/example_sql_execute_query.py[源代码]

execute_query = SQLExecuteQueryOperator(
    task_id="execute_query",
    sql=f"SELECT 1; SELECT * FROM {AIRFLOW_DB_METADATA_TABLE} LIMIT 1;",
    split_statements=True,
    return_last=False,
)

此外，如果您需要使用 Amazon Athena 执行简单的数据质量测试，可以使用 SQLTableCheckOperator

以下示例演示如何实例化 SQLTableCheckOperator 任务。

tests/system/common/sql/example_sql_column_table_check.py[源代码]

row_count_check = SQLTableCheckOperator(
    task_id="row_count_check",
    table=AIRFLOW_DB_METADATA_TABLE,
    checks={
        "row_count_check": {
            "check_statement": "COUNT(*) = 1",
        }
    },
)

参考¶

PyAthena