Airflow PostgresOperator :使用Airflow“redshift”时任务失败，出现异常
Airflow是一个开源的任务调度和工作流管理平台，PostgresOperator是Airflow中的一个操作符，用于执行PostgreSQL数据库的相关操作。当使用Airflow的"redshift"时，如果任务失败并出现异常，可能是由以下几个原因导致：
- 连接错误：检查Airflow配置文件中的数据库连接信息是否正确，包括主机名、端口号、用户名、密码等。
- 数据库权限问题：确保Airflow用户拥有执行相关操作所需的权限，例如查询、插入、更新、删除等。
- 数据库表或视图不存在：确认所需的表或视图是否已在数据库中创建，并且在任务中使用正确的表或视图名称。
- 依赖项缺失：如果任务依赖于其他任务或插件，确保这些依赖项已正确安装和配置。
- SQL语法错误：检查任务中的SQL语句是否符合PostgreSQL的语法规则，包括正确的关键字、表名、列名和运算符等。
针对Airflow和PostgreSQL的相关问题，腾讯云提供了多个产品和服务来满足用户需求：
- 云数据库PostgreSQL：腾讯云提供了高性能、高可用性的云数据库服务，可直接在腾讯云平台上创建和管理PostgreSQL数据库。具体产品介绍和链接地址请参考：云数据库PostgreSQL
- 云服务器CVM：腾讯云提供了云服务器实例，可用于部署Airflow和相关应用程序。用户可以根据需求选择不同规格的云服务器，并在上面自行配置和管理软件环境。具体产品介绍和链接地址请参考：云服务器CVM
- 云函数SCF：腾讯云提供了无服务器计算服务，用户可以使用云函数SCF来编写和运行Airflow的任务代码，无需关心服务器的运维和扩展。具体产品介绍和链接地址请参考：云函数SCF
请注意，以上仅是腾讯云在云计算领域的一些解决方案，其他厂商也提供类似的产品和服务。根据具体需求和情况，可以选择最合适的解决方案来解决Airflow任务失败的问题。
相关·内容
~$ airflow版本2.1.2 python 3.8 我正在尝试使用dag在我的redshift集群上执行一些基本查询，但任务失败并出现异常(日志中未显示) import datetime
import logging
from airflow import DAG
from airflow.contrib.hooks.aws_hook import AwsHook
from airflow.hooks.postgres_hook import PostgresHook
from airflow.operators.postgres_operator import PostgresOp
浏览 119提问于2021-09-17得票数 0
我有一个像下面这样的风景：
只有在源表中为Task 1和Task 2提供新数据时才触发它们(雅典娜)。Task1和Task2的触发器应该在一天内发生新的数据解析时发生。
只在完成Task 3和Task 1时触发Task 2
触发器Task 4只完成Task 3
我的代码
from airflow import DAG
from airflow.contrib.sensors.aws_glue_catalog_partition_sensor import AwsGlueCatalogPartitionSensor
from datetime import dat
浏览 1提问于2020-04-16得票数 5
回答已采纳
1回答
由于某些原因，宏'{{ ds }}‘不递增，执行日期递增，但它不适用于宏出于某种原因，我尝试了一个干净的新dag id，但仍然没有帮助。你知道吗？
from airflow import DAG
from datetime import datetime,timedelta
from airflow.operators.postgres_operator import PostgresOperator
from airflow.sensors.external_task_sensor import ExternalTaskSensor
import sys
sys.path.inser
浏览 1提问于2020-04-20得票数 0
我不知道如何在气流操作员中调用存储过程。存储过程 is- Call goal.dba.sp_ctrl_calendar()
我的密码-
from airflow.operators.postgres_operator import PostgresOperator
sp_ctrl_calendar = PostgresOperator(
task_id = 'sp_ctrl_calendar',
sql = 'goal.dba.sp_ctrl_calendar(); end;',
postgres_conn_id
浏览 3提问于2022-01-06得票数 -1
回答已采纳
1回答
上一次我忘记包括code样本时重新发布。
我在试图设置气流时遇到了一些问题。我一直试图使用一个简单的SQL脚本来完成一个非常基本的DAG设置。我遇到了三个问题：
我能够获得一个基本的DAG运行(我可以在GUI中看到DAG及其运行对数据库的影响)，但似乎DAG不尊重schedule_interval参数。我试图让DAG每小时运行一次，但无论我做了什么更改，DAG都按每天的计划运行，这也反映在我从GUI中看到的情况中。
此外，尽管如果将DAG作为计划运行的一部分运行，DAG能够顺利地运行其单个任务，但我在手动执行DAG时所做的每一次尝试都失败了。我不知道为什么会这样。
在更新DAG s
浏览 0提问于2018-07-27得票数 1
回答已采纳
当尝试在Jinja2中使用气流的模板功能(通过PostgresOperator )时，我无法获得渲染的东西。我很有可能做错了什么，但我对问题可能是什么感到迷茫。下面是一个重现我所得到的TemplateNotFound错误的例子：
airflow.cfg
airflow_home = /home/gregreda/airflow
dags_folder = /home/gregreda/airflow/dags
相关数据集与变量
default_args = {
'owner': 'gregreda',
'start_date':
浏览 3提问于2016-06-14得票数 13
回答已采纳
我有一个python Parent Job和DAG Child Job。Child Job中的任务应该在成功完成每天运行的Parent Job任务时触发。如何添加外部作业触发器？
我的代码
from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.postgres_operator import PostgresOperator
from utils import FAILURE_EMAILS
yesterday = datetime.combine(datetime.tod
浏览 7提问于2020-04-30得票数 22
回答已采纳
我使用的是PostgresOperator，我想把表名后缀传递给我的SQL查询，这样当它查询数据时，它就会动态地从for循环迭代中读取数据。 for country in countries:
matchTimeStamp = ShortCircuitOperator(task_id='Match_Updated_dates_{}'.format(country), provide_context=True,
python_callable=match_dates,op_kwargs={'k
浏览 35提问于2021-08-13得票数 0
回答已采纳
我已经创建了一个红移连接在气流中，附在屏幕截图中。在此之后，我将RedshiftToS3Operator导入到DAG中以运行红移查询，并将csv存储在s3中。
from datetime import timedelta, datetime
import pytz
import airflow
from airflow import DAG
from airflow.operators.dummy import DummyOperator
from airflow.providers.amazon.aws.transfers.redshift_to_s3 import Redshift
浏览 9提问于2022-03-29得票数 1
回答已采纳
我正在使用airflow调度一些任务，当上游任务失败时，我需要下游任务发送错误邮件通知。我使用下面的示例进行测试。
dag_alert_task_callback = DAG(
'dag_alert_task_callback',
default_args=default_args,
schedule_interval=None
)
t1 = PostgresOperator(task_id='create_schema',
sql="CREATE SCHEMA IF NOT EXIS
浏览 0提问于2021-05-14得票数 0
我正在尝试从airflow包中导入PostgresOperator： from airflow.providers.postgres.operators.postgres import PostgresOperator 但是我得到了以下错误：Cannot find reference 'postgres' in imported module airflow.providers。
浏览 42提问于2021-01-09得票数 1
我正在尝试使用一个PythonOperator来执行DAG。当我尝试测试它时，它工作得很好，当我在没有CeleryExecutor的情况下尝试气流时，它也工作得很好。
但当我尝试在使用CeleryExecutor运行的Airflow上回填它时，它失败了，没有真正的描述性错误：
airflow@ip:/home/admin$ airflow backfill REDSHIFT3 -s 2017-05-10
[2017-05-22 14:41:14,373] {__init__.py:57} INFO - Using executor CeleryExecutor
[2017-05-22 14:4
浏览 0提问于2017-05-22得票数 3
我想做一个松弛的警报(失败状态和成功状态)
Dag现在正在工作，当状态为成功状态时，则松弛警报正在工作！但现在，on_failure_callback - fail状态为连续工作(每1分钟一次)
请注意，它仍然失败。但是，我认为这不是有效的，也不是真正的状态。
我怎样才能改变呢？...我想知道实际失败状态的延迟通知
现在我们的task_default参数是这样的。
dt = datetime.now(tz=tz.tzlocal())
task_default_args = {
'owner': 'owner',
'retries'
浏览 2提问于2018-11-30得票数 1
1回答
我在一些ETL中使用气流，在某些阶段，我希望使用临时表(主要是为了使代码和数据对象保持独立，并避免使用大量元数据表)。
使用气流中的Postgres连接和" PostgresOperator“，我发现的行为是:对于，每次执行PostgresOperator，我们在数据库中有一个新的连接(或会话)。换句话说:我们丢失了DAG上一个组件的所有临时对象。
为了模拟一个简单的示例，我使用了以下代码(不要运行，只需查看对象)：
import os
from airflow import DAG
from airflow.operators.postgres_operator import Pos
浏览 0提问于2018-06-14得票数 2
回答已采纳
我有一个sql文件，我想使用PostGresOperator将参数传递给这个sql文件。 """select * from table_{} where id > ID """.format(mytable,myID) 我的postGresOperator mport_redshift_table = PostgresOperator(
task_id='copy_data_from_redshift_{}'.format(country),
postgres_conn_id
浏览 44提问于2021-08-29得票数 0
回答已采纳
1回答
我是第一次接触气流。我想用气流操作符做一个类似下面的操作。 简而言之，我希望从数据库表中读取一些数据，并根据该表中某列的值执行不同的任务。 这是我用来获取数据的表。 +-----------+--------+
| task_name | status |
+-----------+--------+
| a | 1 |
| b | 2 |
| c | 4 |
| d | 3 |
| e | 4 |
+-----------+--------+ 从上表中，我希望选
浏览 19提问于2020-10-06得票数 1
回答已采纳
我是Apache Airflow的新手，我想编写一个DAG来将源数据库中的一组表中的一些数据移动到目标数据库中的一组表中。我正在尝试设计DAG，以便某人可以简单地为新的源表-->目标表进程编写create table和insert into脚本，并将它们放入文件夹中。然后，在下一次运行DAG时，DAG将从文件夹中选取脚本并运行新任务。我将我的DAG设置为：
source_data_check_task_1 (Check Operator or ValueCheckOperator)
source_data_check_task_2 (Check Operator or ValueChec
浏览 7提问于2018-08-06得票数 1
回答已采纳
我有一个DAG，我想用它来回填我的数据库表。
from airflow import DAG
from airflow.operators.postgres_operator import PostgresOperator
from datetime import datetime, timedelta
args = {
'owner': 'airflow',
'depends_on_past': False,
'start_date': datetime(2018, 4, 1),
're
浏览 1提问于2018-05-08得票数 1
我试图以一种动态的方式传递postgres操作符中的参数。
为了刷新元数据，有两个任务，
id (get_query_id_task)的获取列表
传递用于获取和执行查询的ids列表( get_query_text_task)
get_query_id_task = PythonOperator( task_id='get_query_id‘、python_callable=query_and_push、#provide_context=True、op_kwargs={ 'sql’：read_sql(‘warmupqueryid.sql’}) get_query_text_tas
浏览 4提问于2021-05-21得票数 1
回答已采纳
我正在编写一个带有BranchPythonOperator的DAG，以检查是否有数据可供下载。如果数据在那里，DAG应该下载并将其合并到我的PostgreSQL数据库中。如果不存在，则应该跳过所有处理任务，并且分支应该转到DummyOperator。不幸的是，DAG并没有跳过所有的任务。它将跳过最多6个任务，但随后停止(下游任务状态未知)，DAG失败。我在日志中找不到任何错误消息(因为任务没有失败)。
Airflow版本1.8.1。我在下面附上了一些截图。在下面的DAG示例中，我用'XXXXX‘替换了敏感文件信息。我也尝试过ShortCircuitOperator，但只能跳过上海合作组
浏览 46提问于2018-01-04得票数 1
回答已采纳
我有启用VPN连接，Google (PostgreSQL)数据库实例具有相同的VPN连接和SSL启用。云SQL既有公共IP地址，也有私有IP地址。公共IP i用于从外部连接数据库，如PgAdmin客户端工具，以及用于内部连接的私有IP (如数据流)。现在我想连接这个来自Composer的CloudSQL。使用PostgresOperator连接Cloud数据库。在气流->连接部分创建与Puplic作为端口的单独连接。由于此CloudSQL启用了SSL，所以将证书推送到DAG的GCS位置。在刚才传递ssl证书路径信息的额外属性部分下的连接中，如下所示，
{
"sslmode&
浏览 6提问于2019-09-17得票数 4
我想使用执行日期作为我的sql文件的参数：
我试过了
dt = '{{ ds }}'
s3_to_redshift = PostgresOperator(
task_id='s3_to_redshift',
postgres_conn_id='redshift',
sql='s3_to_redshift.sql',
params={'file': dt},
dag=dag
)
但它不起作用。
浏览 4提问于2017-06-13得票数 15
回答已采纳
我希望将查询“从<table_name>中选择计数(*)”的结果从PostgresOperator中推送到PythonOperator，后者是pythonCallable，该值将用于python函数。我如何在其中使用XCom？我尝试过不同的方法，但每次我在python函数中得到一个未定义的变量时都会出错。
def <python_function_name>():
"""variable to be used in this code."""
python_airflow_step = PythonOperator(
浏览 5提问于2022-07-13得票数 0
我正在尝试创建一个气流管道，用于从API下载数据，处理数据，将其保存为CSV，然后将数据加载到Postgres数据库(所有数据都在码头容器中)。代码看起来像这样
from datetime import datetime, timedelta
import pandas as pd
from airflow import DAG
from airflow.providers.postgres.operators.postgres import PostgresOperator
from airflow.operators.python import PythonOperator
defaul
浏览 10提问于2022-01-27得票数 1
2回答
我正在尝试编写一个管道，postgres db应该使用csv的内容进行更新，当它被带到文件夹时。我已经编写了一个dag，它创建了表，并在从web中触发csv内容时推送它。下面是代码：
from datetime import datetime
from airflow import DAG
from airflow.utils.trigger_rule import TriggerRule
from airflow.operators.postgres_operator import PostgresOperator
from airflow.operators.python_operator
浏览 3提问于2020-11-26得票数 4
回答已采纳
我使用Ubuntu在运行Win 10的本地机器上设置了气流2.0。我使用PostgreSQL作为数据库，CeleryExecutor和RabbitMQ作为芹菜后端。我创建了一些DAG，每个DAG都通过SSH隧道连接到Redshift数据库，并执行SQL命令。当我手动触发或通过调度程序运行时，每个DAG都会顺利运行。
但是，当我为这些DAG同时开始运行设置日程时，我遇到了一个错误。例如，如果DAG1和DAG2在上午8:00开始运行，这2个dags将失败并显示如下错误：
or :服务器意外地关闭了连接，这可能意味着服务器在处理请求之前或处理过程中异常终止。
如果我在不同的时间设置这两个跳闸，一切
浏览 11提问于2021-01-12得票数 0
回答已采纳
我使用的是PostgresSQL操作符。任务如下：
emailage_transformations = PostgresOperator(
task_id = 'emailage_transformations',
sql = '/home/ubuntu/airflow_ci/current/scripts/antifraud/emailage_transformations.sql',
postgres_conn_id = 'redshift',
autocommit = True,
dag = dag
浏览 2提问于2019-09-05得票数 4
回答已采纳
1回答
我正在学习Airflow，我正在尝试理解连接是如何工作的。
我有一个包含以下代码的第一个dag：
c = Connection(
conn_id='aws_credentials',
conn_type='Amazon Web Services',
login='xxxxxxxx',
password='xxxxxxxxx'
)
def list_keys():
hook = S3Hook(aws_conn_id=c.conn_id)
logging.info(f"Lis
浏览 13提问于2020-05-22得票数 2
回答已采纳
我试图将执行日期作为运行时参数传递给postgres操作符。
class MyPostgresOperator(PostgresOperator):
template_fields = ('sql','parameters')
task = MyPostgresOperator(
task_id='test_date',
postgres_conn_id='redshift',
sql="test_file.sql",
parameters={'crunch_date':&
浏览 1提问于2019-08-09得票数 5
回答已采纳
我正在尝试运行一个气流管道，其中我有一个主要Postgres任务，然后是一个Python循环任务，然后是一个最终postgres任务，所以这个dag类似于如下所示
PythonOperator
|Task B.1|
Postgres |Task B. | Postgres
Task A ------|Task B. |-----> Task C
|Task B.n|
所以，我写了这篇文章：
import glob
import logging
import os
from datetim
浏览 1提问于2021-05-24得票数 1
回答已采纳
我们可以在气流后置参数中传递数组列表吗？
我在尝试像这样的东西
Select hotel_name from hoteldetails where id in ({{ params.hotel_ids }});
在“s3_to_redshift.sql”中。
下面是我的运算符：
s3_to_redshift = PostgresOperator(
task_id='s3_to_redshift',
postgres_conn_id='redshift',
sql='s3_to_redshift.sql',
para
浏览 4提问于2020-02-12得票数 0
因此，我试图在partial()中使用PostgresOperator方法，但是我得到这个错误显然是因为我无意中将schedule_interval传递给它。我查找了，BasicOperator的分部()方法没有这样的参数，我假设它是所有操作符类的父类。
所以我很困惑，我必须把这个参数传递给DAG，但是.partial()没有这样的参数，那么我应该如何创建这个进程和任务呢？我还没有找到任何关于如何实现它的信息。
from airflow import DAG
from airflow.decorators import task
from airflow.providers.postgres.
浏览 4提问于2022-09-12得票数 0
回答已采纳
当出现异常时，我正在尝试使airflow将任务标记为失败。我已经在磁盘上保存了一个python脚本，我希望使用BashOperator运行该脚本，如下所示：
import sys
def myfunc(a):
if a == True:
return
else:
raise Exception
if __name__ == '__main__':
try:
myfunc(False)
except:
sys.exit(1)
当运行这样的代码时，我希望在Graph View中看到一个
浏览 10提问于2018-02-08得票数 1
我创建了一个xcom，我想以PostgresOperator参数的形式获得结果。我试过了 my_task = PostgresOperator(
task_id=‘my_task',
postgres_conn_id=config.get(env, 'redshift_conn'),
sql="my_task.sql",
params={
‘my_parameter': {{ int(ti.xcom_pull(task_ids=‘previous_task')) }}
},
浏览 20提问于2020-10-30得票数 3
回答已采纳
我有一个有近50个任务并行运行的Airflow DAG。偶尔，我会遇到这样的异常。“尝试心跳时出现异常！睡眠5.0秒”。Airflow文档说，这是由于任务的周期性心跳失败，调度器认为他们是僵尸。任何修复都会很有帮助。
浏览 0提问于2018-10-01得票数 1
1回答
我有我的气流项目，结构如下
airflow
|
|----dags
| |----dag.py
|
|----dbt-redshift
|----models
|----model.sql
我在卷部分中包含了dbt-redshift目录，如
volumes:
-./dbt-redshift:/opt/airflow/dbt-redshift
我试图使用bash操作符在DAG中运行dbt
dbt_task = BashOperator(task_id='dbt'
浏览 14提问于2022-05-24得票数 0
回答已采纳
我正在为亚马逊AirFlow 1.10.12创建一个插件和dag结构。我根据文档这样做： dags:
- aws_from_redshift_to_s3.py
plugins:
- __init__.py
- from_redshift_to_s3_plugin.py
- operators:
-- __init__.py
-- aws_from_redshift_to_s3_operator.py aws_from_redshift_to_s3_operator.py： from airflow.hooks.postgres_hook import PostgresHook
浏览 30提问于2021-08-16得票数 0
我试图使用target-redshift将数据推送到aws-redshift
我使用airflow来监视etl状态。
这是错误日志，我不知道它意味着什么。在线文档几乎不存在目标红移。有办法避免这个错误吗？
[2021-03-07 15:04:11,841] {bash_operator.py:126} INFO - ERROR Exception writing records
[2021-03-07 15:04:11,841] {bash_operator.py:126} INFO - Traceback (most recent call last):
[2021-03-07 15:04
浏览 4提问于2021-03-08得票数 3
回答已采纳
我正在使用Airflow来协调ETL。当用户创建没有后期绑定的视图时，许多任务都有失败的drop table子句。不能将所有任务更改为drop cascade。
有没有办法强制进行后期绑定？通过用户权限或Redshift配置。
浏览 19提问于2019-07-08得票数 2
回答已采纳
1回答
几个小时前，这段代码运行良好，但突然之间，我的伙伴们开始陷入“排队”状态。
下面是我试图运行的内容(我只是手动触发)：
from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from datetime import datetime as dt
def test_function():
print('Hello there')
default_args = {
'owner': 'airflow',
&#
浏览 3提问于2022-05-24得票数 0
1回答
我是airflow的新手。我根据官方文件设置了所有的东西。我使用pet example DAG，但是当我查看DAG日志时，它显示了以下错误：
第一错误从populate_pet_table开始
psycopg2.errors.InvalidTextRepresentation: invalid input syntax for type integer: "Maxy"
LINE 2: INSERT INTO pet VALUES ('Maxy', 'Dog', '2018-07-...
第二错误从get_birth
浏览 1提问于2021-09-14得票数 0
回答已采纳
只有将SQL文件放在与DAG相同的文件夹中，我才能运行它：
create_tables = PostgresOperator(
task_id="create_tables",
dag=dag,
postgres_conn_id="redshift",
sql="create_tables.sql"
)
如果create_tables.sql位于不同的文件夹(父文件夹、兄弟文件夹)中，则PostgresOperator抛出jinja2.exceptions.TemplateNotFound，我尝试了绝对路径、相对路径(./createtables
浏览 3提问于2020-01-10得票数 2
回答已采纳
我正在使用气流2.0。我试图连接到红移，使用码头集装箱上的mac。
这是我的dag.py：
from airflow import DAG
from datetime import datetime,timedelta
from airflow.operators.bash import BashOperator
from airflow.operators.python import PythonOperator,BranchPythonOperator
from airflow.providers.jdbc.hooks.jdbc import JdbcHook
#Creating
浏览 20提问于2021-04-01得票数 1
我需要使用copy命令将s3文件复制到红移。我对气流有点陌生，遇到了一些问题。有人能改正下面的代码吗？我可以这样调用rs.execute()吗？ Error:
op.execute()
TypeError: execute() missing 1 required positional argument: 'context' 代码： import os
from airflow import DAG
from airflow.hooks.S3_hook import S3Hook
from airflow.operators.python_operator impor
浏览 26提问于2020-12-18得票数 1
1回答
我发现xcom实际上将数据写入数据库并从其他任务中提取数据。我的数据集很大，对它进行分类和写入数据库会造成一些不必要的延迟。是否有一种不使用xcom在同一气流Dag中的任务之间通信数据的方法？
下面是我尝试过的代码，上下文实际上没有被传递。我知道使用task_instance.xcom_push()可以工作，但它也会对数据进行筛选，并将其写入我不需要的数据库中。
from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from datetime import datetime, t
浏览 4提问于2020-02-18得票数 3
1回答
我想在get_birth_date中使用来自get_all_pets的结果。我如何在get_birth_date中访问它？此外，我还想打印get_all_pets的结果，我在哪里可以看到这样的print()？
在我的代码中哪里可以这样做？
import datetime
from airflow import DAG
from airflow.providers.postgres.operators.postgres import PostgresOperator
# create_pet_table, populate_pet_table, get_all_pets, and get_b
浏览 7提问于2022-05-08得票数 1
我对气流很陌生。我的要求是开发一个数据管道，如下所示：
( i)连接并下载多个csv文件(文件名以名称开头)气流从给定的SFTP文件夹到S3
(2)从S3到redshift的文件
我已经使用了SFTPToS3Operator和S3ToRedshiftOperator来完成这个任务，当我为sftp_path和s3_key提供完整的文件名时，它可以很好地工作，但是如果我在sftp_path和s3_key中为文件名提供了通配符(airflow*.csv)，就会出现IO错误。
需要您对从SFTP下载多个文件到s3并摄入红移的方法的专家意见。
我正在为单个文件使用下面的代码，该代码对单个文件运行良好。
浏览 8提问于2021-12-24得票数 2
首先，我连接了气流PostgresOperator并创建了数据集，现在我想将它发送到GCS操作符连接，但是总是看到我无效的dsn:无效连接选项"extra__google_cloud_platform__key_path“。
我怎么换衣服？？
我的气流连接
还有我的达格斯
postgres_to_gcs_task = PostgresToGCSOperator(
task_id = 'postgres_to_gcs_task',
postgres_conn_id='postgres_to_gcs',
sql = 'SELE
浏览 6提问于2022-09-08得票数 0
使用airflow，我从MySQL数据库中提取数据，使用python进行转换，并将其加载到Redshift集群中。
目前我使用了3个airflow任务:它们通过在本地磁盘上写入CSV来传递数据。
在不写入磁盘的情况下如何做到这一点？我应该用python写一个大任务吗？(这会降低可见性)
编辑:这是一个关于气流的问题，是选择任务粒度以及如何在它们之间传递数据的最佳实践。这不是一个关于数据迁移或ETL的一般性问题。在本问题中，ETL仅用作气流任务的工作负载示例。
浏览 36提问于2017-01-10得票数 13
我试图使用气流运行一个查询在雅典娜使用气流在一个时间表。
下面附带的函数是使用PythonOperator在airflow中运行的。
from airflow.models import Variable
from airflow.contrib.hooks.aws_athena_hook import AWSAthenaHook
import datetime
def update_athena_partition(*args, **kwargs):
execution_date = datetime.datetime.strptime(kwargs['ds'],
浏览 1提问于2019-08-13得票数 2
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
即时通信 IM
实时音视频活动推荐
腾讯云开发者
