Amazon Athena SQL¶

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

先决条件任务¶

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

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

操作符¶

执行 SQL 查询¶

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

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

tests/system/providers/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/providers/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