如何在另一个任务airflow中使用查询结果(bigquery运算符

文章/答案/技术大牛

发布

1回答

Airflow - Bigquery运算符未按预期工作

sql、google-bigquery、airflow

我对Airflow是个新手，目前我被Bigquery操作符的问题所困扰。我正在尝试对给定数据集中的一个表执行一个简单的查询，然后将结果复制到同一数据集中的一个新表上。我使用bigquery运算符来做这件事，因为根据文档，'destination_dataset_table‘参数应该做我正在寻找的事情(来源：https://airflow.apache.org/docs编辑:我在一个小得多的表(从10 Kb到几Kb

浏览 7提问于2020-09-11得票数 0

1回答

用于在Salesforce表中运行查询的自定义airflow操作符

salesforce、airflow、custom-operator

我是airflow的新手，我想安排一个任务，让来自不同数据库记录计数的两个表必须检查它是否匹配。一个来源是GCP，另一个是Salesforce。所以我找到了BigQueryOperator来点击GCP端的查询并返回计数结果，但是我找不到任何看起来像SalesforceQueryOperator的运算符，我可以在气流任务中分配它。所以基本上，我说的是这个，我们可以用它来得到计数结果： task

浏览 7提问于2021-02-12得票数 1

1回答

如何利用局部bq算子从气流中实现python任务

google-bigquery、airflow、bq

在进行项目时，遇到了这样一种情况:我们希望使用气流来执行很少的任务，但是不允许使用python操作符，而是被指示使用本地的BigQuery操作符。谁能帮我这样的代码，如设置气流变量，或如何编写条件代码，这将执行by操作。有可能吗？如果没有，那么我的下一个问题是，是否有可能使用BQ操作符从BQ表中获取结果，并将其分配给一个python变量，所以想同时使用BQ和python操作符，有什么方法吗？

浏览 6提问于2021-11-28得票数 1

回答已采纳

1回答

airflow、google-cloud-composer

} || bq mk {}'.format( >> delete_bq_table export_audits_to_gcs >> download_file >> email_summary 使用此代码然后，我将.

浏览 18提问于2021-08-19得票数 0

回答已采纳

2回答

添加带有gdrive作用域的BigQuery连接？

google-cloud-composer

我有一个外部Sheets表，我想通过Airflow中的BigQueryOperator查询它。Conn Id: bigquery_with_gdrive_scopeProject,https://www.googleapis.com/auth/cloud-platfor

浏览 4提问于2018-08-09得票数 2

回答已采纳

1回答

是否可以使用BigqueryOperator在一次调用中将数据结果存储到多个分区中？

airflow

query_to_table_x.sql SELECT column_a, column_b, column_cWHERE _PARTITIONTIME = {{ execution_date }} 我有airflow_dag.py示例:今天是2020-04-06，所以我将使用filter _PARTITIONTIME = '2020-04-06'运行run query_to_table_x.sql，然后将结果存储到project.dataset.result_table_x$20200406

浏览 11提问于2020-04-07得票数 0

1回答

GCP数据摄取体系结构

google-cloud-platform、google-cloud-dataflow

我将开始研究从CSV和数据湖中获取到BigQuery的GCP数据，我正在寻求您的建议，我可以使用哪些技术或架构。我是GCP的新手，但我对数据工程技术有很好的理解，因为我正在为认证做准备。我尝试使用转换成管道的数据流简单地加载BigQuery，然后调度器在特定的时间运行管道。我遇到的问题是如何将已处理的CSV文件移动到其他桶中？我可以调用函数来移动文件吗？

浏览 4提问于2022-10-21得票数 1

1回答

google-cloud-composer BigQuery交叉数据集加载

google-bigquery、google-cloud-composer、airflow

我正在尝试运行一个简单的select查询(从biqquery)，并使用Composer将结果集加载到另一个bq表中。然而，我在代码的最后一行得到了一个错误。from airflow import DAGfrom airflow.contrib.operatorsimport bigquery_operator from

浏览 1提问于2021-04-09得票数 0

2回答

如果表中已存在内容，则避免在Bigquery中重新加载数据

google-bigquery、airflow、etl

我需要编写dag，这样它就可以将csv文件的内容加载到Bigquery表中，但是如果表中已经存在该内容，它就会丢弃该文件。我知道我们可以在正常的编程中计算文件的哈希，但不知道如何处理Airflow和Bigquery。有人知道怎么做吗？

浏览 0提问于2021-09-13得票数 0

2回答

AssertionError: INTERNAL:未指定默认项目

python-2.7、airflow、apache-airflow

不熟悉airflow。尝试运行sql并将结果存储在BigQuery表中。请帮帮我。/contrib/operators/bigquery_operator.py", line 82, in execute File "/usr/

浏览 12提问于2017-08-11得票数 4

回答已采纳

2回答

是否有基于作业输出发送通知的api？

qubole

但是，假设我运行一个hive查询来计算表中的行数，会发生什么呢？如果返回的结果是零，我想发送电子邮件给有关方面。我该怎么做呢？谢谢。

浏览 5提问于2016-01-29得票数 0

5回答

使用Airflow在数据库之间移动和转换数据

airflow

使用airflow，我从MySQL数据库中提取数据，使用python进行转换，并将其加载到Redshift集群中。编辑:这是一个关于气流的问题，是选择任务粒度以及如何在它们之间传递数据的最佳实践。这不是一个关于数据迁移或ETL的一般性问题。在本问题中，ETL仅用作气流任务</

浏览 36提问于2017-01-10得票数 13

1回答

我们能在通过Google BigQuery运行的查询上实现数据谱系吗？

google-cloud-platform、google-bigquery、airflow、google-cloud-composer

谁能帮我提供一些建议，告诉我如何在基于谷歌BigQuery的DW类型解决方案上实现数据谱系，使用谷歌云存储作为源，使用谷歌云组合器作为工作流管理器来实现一系列SQL。

浏览 0提问于2018-10-15得票数 0

5回答

从气流中的BigQueryOperator中提取结果

python、google-bigquery、airflow

我试着用气流从BigQueryOperator中获取结果，但是我找不到方法去做。我尝试在next()成员(在1.10中可用)中调用bq_cursor方法，但是它返回None。我就是这么做的import logging from airflow.contrib.operators importbigquery_operator from airflow.operat

浏览 0提问于2018-11-30得票数 5

回答已采纳

1回答

在气流中进行数据库请求时出现的问题

airflow

我正在尝试基于数据库调用的响应动态创建任务。但是当我这样做时，run选项就不能进入气流，所以我不能运行。task_id = "materialize_" + vname, google_cloud_storage_conn_id="google

浏览 0提问于2019-08-29得票数 0

3回答

Airflow:有没有办法在dag之外将操作员分组在一起？

python、airflow、google-cloud-composer

然后尝试在Big Query中加载原始表中的数据，然后根据加载结果将文件分派到存档或拒绝的文件夹中。做一次很容易，如果需要做1000次呢？我正在试图弄清楚如何优化工程时间。可以考虑使用SubDag，但它在performances方面存在局限性，将被弃用anyway。任务组需要是dag的一部分才能实现https://github.com/apache/airflow/blob/1be3ef635fab635f741b775c52

浏览 39提问于2021-09-06得票数 1

回答已采纳

4回答

气流BigQueryOperator:如何在分区表中保存查询结果？

google-bigquery、airflow

我有一个简单的DAGfrom airflow.contrib.operators.bigquery_operator import BigQueryOperatorCREATE_IF_NEEDED', start >> bq_query >> end 执行bq_query任务时，SQL查询将保存在切分表中。将

浏览 2提问于2018-05-24得票数 6

1回答

使用Airflow调度数千个任务

airflow、airflow-scheduler

我们正在考虑将Airflow用于一个项目，该项目每天需要对外部API进行数千次调用以下载外部数据，每次调用可能需要几分钟的时间。我们正在考虑的一个选择是为每个不同的API调用创建一个任务，但是这将导致数千个任务。在UI中渲染所有这些任务将是一个挑战。我们还担心调度器，它可能会处理如此多的任务。另一种选择是只有几个并行的长时间运行的任务，然后在这些任务中实现我们自己的调度程序。我们可以在PythonOperator中添

浏览 6提问于2020-03-07得票数 1

1回答

如何使用JDBC操作符在airflow中获取sql查询结果

jdbc、airflow

我已经在Airflow connections中配置了JDBC连接。我的DAG的任务部分如下所示，其中包含一条select语句。当触发DAG成功时，查询结果不会打印在日志中。如何使用JDBC运算符获取查询结果。

浏览 41提问于2021-01-05得票数 0

2回答

如何使BigQuery* SQL管道自动化*

google-cloud-platform、google-bigquery、google-cloud-dataflow

我使用BigQuery SQL创建了一个数据管道。首先从云存储中导入CSV文件，然后进行不同的分析，包括利用BigQueryML地理函数进行预测建模和利用解析函数进行KPI计算。我能够成功地手动运行不同的查询，现在我想要自动化数据管道。我的第一个选择是DataFlow SQL，但事实证明Dataflow SQL查询语法不支持地理函数。

浏览 3提问于2020-04-21得票数 2

回答已采纳

点击加载更多

Airflow - Bigquery运算符未按预期工作

用于在Salesforce表中运行查询的自定义airflow操作符

如何利用局部bq算子从气流中实现python任务