有没有办法在 Python 运算符中读取 MWAA 上气流 Dag 的 SQL 文件？答案

【问题标题】：Is there a way to read in an SQL file for an airflow Dag on MWAA in a Python Operator?有没有办法在 Python 运算符中读取 MWAA 上气流 Dag 的 SQL 文件？
【发布时间】：2022-12-30 01:44:32
【问题描述】：

我试图通过读取包含我要运行的查询的 SQL 文件来简化我的工作流程。在 Python 运算符中有一个雪花运算符，它被执行，然后所有这些信息都被传递到一个 API，该 API 被设置为使用来自 SQL 查询的数据。所有 DAG 工作流都通过 MWAA 在 AWS 中进行管理，但每当我尝试读取包含在 S3 文件夹中的包含所有 dag 的 SQL 文件时，它都会指出找不到该文件。所以我的问题基本上是，有没有一种方法可以直接使用 python 的 open 函数读取文件，或者我是否必须拆分 DAG 任务以使 Snowflake 运算符与 python 运算符分开？

【问题讨论】：

您能否提供更多详细信息，例如您如何尝试从 S3 读取文件，或者您是否可以在此处显示示例代码。

标签： python directed-acyclic-graphs mwaa

【解决方案1】：

是的，您可以通过操作员中的 open 读取存储在 s3 中的文件，但您不需要这样做。相反，您可以将 sql 放在与包含 DAG 文件的目录相关的位置。因此，如果您的 DAG 文件位于 /usr/local/airflow/dags/test_dag.py，您可以将 test.sql 文件移动到 /usr/local/airflow/dags/ 下的任何位置（示例：/usr/local/airflow /dags/query/test.sh) 并将相对路径传递给参数，如下所示：

t2 = MyOperator(
task_id="example",
# "scripts" folder is under "/usr/local/airflow/dags"
sql_query="query/test.sh",
dag=dag,

)

【讨论】：