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 查询。

要对 Amazon Athena 执行单个 SQL 查询，而无需将结果返回到 Airflow，请改用 AthenaOperator 代替。

common/sql/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 任务。

common/sql/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