我必须向bigquery表添加标签。我知道通过BigQuery UI做它，但它如何可以通过气流操作。
用例：用于计费和搜索目的。当多个团队在同一个项目和数据集下工作时，我们需要将各个团队创建的所有表合并在一起。因为每个团队对表都有不同的标签，所以标签对我们来说是必要的。
bq_query = BigQueryOperator(bql=sql,
destination_dataset_table='my_dataset.my_table'),
task_id='
浏览 2提问于2021-11-16得票数 0
回答已采纳
我是Apache Airflow的新手。我的任务是从谷歌云存储读取数据，转换数据并将转换后的数据上传到BigQuery表中。我可以从云存储存储桶中获取数据，并直接存储到BigQuery表中。我不确定如何在这个管道中包含transform函数。 下面是我的代码： # Import libraries needed for the operation
import airflow
from datetime import timedelta, datetime
from airflow import DAG
from airflow.operators.dummy_operator import
浏览 18提问于2021-07-09得票数 1
回答已采纳
我已经创建了下面的DAG用于每天运行sql脚本。如何将查询结果保存为JSON文件并保存到Google Composer的DAG文件夹中？ import datetime
import airflow
from airflow.operators import bash_operator
from airflow.contrib.operators import bigquery_operator
START_DATE = datetime.datetime(2020, 3, 1)
default_args = {
'owner': 'Alen',
浏览 9提问于2020-03-18得票数 0
我正在尝试使用bigquery钩子在airflow中创建bigquery视图，并给出以下错误。请找到下面的代码片段： BigQueryHook().create_empty_table(
dataset_id="data_loader_view",
table_id="customerdata",
view={
"query": "SELE
浏览 14提问于2020-10-01得票数 1
回答已采纳
我正在尝试运行一个简单的select查询(从biqquery)，并使用Composer将结果集加载到另一个bq表中。然而，我在代码的最后一行得到了一个错误。
损坏的DAG: /home/airflow/gcs/dags/es_tc_etl_wkf_mtly.py无效语法(es_tc_etl_wkf_mtly.py，第47行)
代码：
import airflow
from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from airflow.contrib.operators impor
浏览 1提问于2021-04-09得票数 0
1回答
我正在尝试为GCS桶上存在的Parquet文件在Big Query中创建一个外部表。但是，我在气流中运行下面的代码时出错了：
错误：
ERROR - 400 POST https://bigquery.googleapis.com/bigquery/v2/projects/project_dev/datasets/dataset_dev/tables?prettyPrint=false: When defining a table with an ExternalDataConfiguration, a schema must be present on either the Table or
浏览 2提问于2021-10-20得票数 0
回答已采纳
我是airflow的新手，有一个关于Airflow及其处理器的问题。当处理器产生输出时，如何将此输出移到下一个处理器的输入中？有一个软件叫做nifi，它将中间输出存储到flowfile中，afaik在airflow中没有这样的东西。那么这是如何发生的呢？
提前谢谢。
浏览 14提问于2017-03-13得票数 10
回答已采纳
情况是，当我执行这个查询(select query from `bigquery-analytics-workbench.team_bi._airflow_logs_tests)时，我得到了下一个结果：
我想执行保存在此列中的查询，但是当我执行此查询时，EXECUTE IMMEDIATE select query from bigquery-analytics-workbench.team_bi._airflow_logs_tests limit 1返回错误Not found: Dataset bigquery-analytics-workbench:team_bi was not fo
浏览 10提问于2021-05-19得票数 0
我在这里(https://airflow.apache.org/docs/apache-airflow-providers-google/stable/_modules/airflow/providers/google/cloud/operators/bigquery.html#BigQueryGetDataOperator)阅读了Airflow的作业操作员文档，但我找不到如何将作业优先级更改为BigQuery。如何做到这一点呢？
浏览 26提问于2021-06-23得票数 0
回答已采纳
我对Airflow是个新手，目前我被Bigquery操作符的问题所困扰。我正在尝试对给定数据集中的一个表执行一个简单的查询，然后将结果复制到同一数据集中的一个新表上。我使用bigquery运算符来做这件事，因为根据文档，'destination_dataset_table‘参数应该做我正在寻找的事情(来源：https://airflow.apache.org/docs/stable/_api/airflow/contrib/operators/bigquery_operator/index.html#airflow.contrib.operators.bigquery_operato
浏览 7提问于2020-09-11得票数 0
我正在写一个airfllow管道。我需要编写dag，这样它就可以将csv文件的内容加载到Bigquery表中，但是如果表中已经存在该内容，它就会丢弃该文件。我知道我们可以在正常的编程中计算文件的哈希，但不知道如何处理Airflow和Bigquery。有人知道怎么做吗？
浏览 0提问于2021-09-13得票数 0
我试图通过airflow (安装在Kubernetes中)将表交叉点的结果写入csv文件，但我得到的消息是csv文件是一个读文件。在脚本中有没有我可以修改的参数，让它写出结果？ def update_table():
# Connecting to BigQuery
client = bigquery.Client()
query = """SELECT ltrim(rtrim(col1)) as col1,
sum(col2) as col2
FROM dataset.tab
浏览 13提问于2019-10-15得票数 0
我有一个简单的DAG
from airflow import DAG
from airflow.contrib.operators.bigquery_operator import BigQueryOperator
with DAG(dag_id='my_dags.my_dag') as dag:
start = DummyOperator(task_id='start')
end = DummyOperator(task_id='end')
sql = """
SE
浏览 2提问于2018-05-24得票数 6
我尝试使用cloud composer来调度和编排Bigquery作业。Bigquery表在澳大利亚-东南1 region.The云composer环境是在美国中央1地区创建的(因为composer在澳大利亚地区不可用)。当我尝试下面的命令时，它抛出一个模糊的错误。当我尝试使用驻留在欧盟和美国的数据集时，相同的设置工作得很好。 Command:
gcloud beta composer environments run bq-schedule --location us-central1 test -- my_bigquery_dag input_gl 8-02-2018
Error:
浏览 11提问于2019-02-13得票数 0
回答已采纳
我在GCP composer上有一个每5分钟运行一次的Airflow DAG。我想创建一个BigQuery表，其中包含DAG开始运行的时间和一个标识它是成功运行还是失败运行的标志。例如，如果DAG在2020-03-23 02:30运行，但运行失败，则BigQuery表将具有time列2020-03-23 02:30和flag列1。如果运行成功，则表将具有time列2020-03-23 02:30和flag列0。该表将追加新行。 提前感谢
浏览 23提问于2020-03-24得票数 0
回答已采纳
我正在尝试设置数据管道，将数据从GCS移动到BigQuery，执行某些任务/处理并将其加载到MongoDB集群(所有这些都是在python中使用DAG设置的)。我一直能够做到这一点，直到加载到MongoDB。是否有任何现有的气流操作人员能够做到这一点？如果没有，是否可以使用气流中提供的mongoDB挂钩创建自定义代码？
谢谢，GT
编辑1
我使用了MongoHook和BigQueryGetDataOperator的源代码(下面是代码片段)。我现在的问题是，我需要对10++百万行执行此操作，并且当我在BigQueryGetDataOperator中增加max_results='100&#
浏览 2提问于2018-11-20得票数 1
我目前正在使用Airflow PostgresToGoogleCloudStorageOperator和GoogleCloudStorageToBigQueryOperator将我的Postgres DB (托管在AWS RDS上)的每个表导出到BigQuery。它可以工作，但我有75张桌子，所以Airflow创造了75 *2个工作岗位。由于我刚接触Airflow，我不知道这是不是一个好的做法。 在任何情况下，我都希望找到一种一次性导出所有表的方法(pg_dump?)转换为GCS，然后将其导入BigQuery。
浏览 13提问于2020-01-31得票数 5
回答已采纳
我正在尝试将一个表从BigQuery导出到Google数据库。
我找到了一个叫BigQueryToMySqlOperator的操作符(在这里有文档，)
当我将包含此任务的DAG部署到composer时，任务总是由于错误而失败。
Traceback (most recent call last):
File "/opt/python3.8/lib/python3.8/site-packages/airflow/models/taskinstance.py", line 1113, in _run_raw_task
self._prepare_and_execute_t
浏览 2提问于2021-07-23得票数 0
回答已采纳
我想从airflow.providers.google.cloud.sensors.bigquery导入BigQueryTableExistenceAsyncSensor
这是我的代码：
from airflow import DAG
from util.dags_hourly import create_dag_write_append #this is class that I created, no issues with other DAG
from airflow.providers.google.cloud.sensors.bigquery import
BigQueryTabl
浏览 12提问于2022-10-25得票数 0
回答已采纳
关于我的Aitflow流程，我有以下结构：
dags/mainDag.py
dags/BigQuery/deleteData.py
dags/BigQuery/insertData.py
dags/support/gcp.json
dags/support/__init__py
我的mainDag.py正在调用deleteData.py和insertData.py，这是可行的！但我的问题是:在这两个文件中，我都像这样使用gcp.json：
credentialsPath = "~/airflow/dags/support/gqp.json"
bigqueryClient = b
浏览 4提问于2020-03-25得票数 1
回答已采纳
不熟悉airflow。尝试运行sql并将结果存储在BigQuery表中。
获取以下错误。不确定在何处设置default_rpoject_id。
请帮帮我。
错误：
Traceback (most recent call last):
File "/usr/local/bin/airflow", line 28, in <module>
args.func(args)
File "/usr/local/lib/python2.7/dist-packages/airflow/bin/cli.py", line 585, in test
浏览 12提问于2017-08-11得票数 4
回答已采纳
我有一个外部Sheets表，我想通过Airflow中的BigQueryOperator查询它。
我更喜欢使用Cloud Composer服务帐户。
我已经通过Airflow UI创建了一个新的连接，参数如下：
Conn Id: bigquery_with_gdrive_scope
Conn Type: google_cloud_platform
Project Id: <my project id>
Keyfile path: <none>
Keyfile JSON: <none>
Scopes: https://www.googleapis.com/aut
浏览 4提问于2018-08-09得票数 2
回答已采纳
我使用Airflow DataflowTemplateOperator()将数据从Mssql迁移到Bigquery，使用JDBC to Bigquery数据流模板。
默认情况下，它会将数据追加到目标Bigquery表中。
我想先截断该表，然后再写入新行。
是否有默认参数可将数据流/ DataflowTemplateOperator设置从追加更改为覆盖？
浏览 2提问于2020-12-02得票数 0
我想收到一个电子邮件通知的任务成功，失败和重试在GCP使用Sendgrid。
目前，我的DAG中的所有任务都在成功运行。在这种情况下，我想收到通知。
另外，当某些任务失败或重新尝试时，我也希望得到这些通知。我执行了以下步骤，并且在强制任务失败时没有收到任何通知。
创建GCP环境，添加环境变量。
SENDGRID_MAIL_FROM : abc@gmail.com
SENDGRID_API_KEY :
在DAG之后创建。
import json
from datetime import timedelta, datetime
from airflow import D
浏览 4提问于2020-04-01得票数 0
回答已采纳
我现在有一段代码，我正试图进入气流中。
该代码从表中获取数据，并在def map_manufacturer_model中预先构造函数，并将其输出到bigquery表中。
我知道代码在我的电脑上工作，但是，当我把它移到气流中时，我得到了错误：
TypeError:缺少一个必需的位置参数：'s'
有人能在这方面提供任何指导吗，因为我现在被困住了！
谢谢
import pandas as pd
import datetime
from airflow.operators import python_operator
from google.cloud import bigquery
浏览 4提问于2021-12-15得票数 0
回答已采纳
我想检查一下我的表是否加载正确。如果它没有正确加载，那么记录的数量将为零。我正在使用SQLCheckOperator来完成此任务。
这是代码
from airflow.operators.sql import SQLCheckOperator
from datetime import date, timedelta
CURRENT_DATE = str(date.today() - timedelta(2))
TABLE_NAME = "foo"
search_monolith_post_sanity = SQLCheckOperator(
task_id="
浏览 0提问于2021-02-02得票数 2
我需要创建一个airflow操作符，它接受几个输入，并返回一个字符串，该字符串将用作下一次运行的另一个操作符的输入。我对airflow dags和操作员是新手，对如何正确地这样做感到困惑。由于我是为使用airflow和构建dag的人构建此应用程序，而我不是一个实际的airflow用户或dag开发人员，因此我想获得有关正确使用它的建议。我已经创建了一个运算符，它返回一个令牌(只是一个字符串，所以hello world运算符示例可以很好地工作)。这样做，我看到了用于dag执行的XCOM值。但是如何正确地检索该值并将其输入到下一个运算符？在我的示例中，我只是调用了相同的运算符，但实际上它将调用一个
浏览 26提问于2021-08-14得票数 0
回答已采纳
我正在尝试按照https://aws.amazon.com/blogs/big-data/migrating-data-from-google-bigquery-to-amazon-s3-using-aws-glue-custom-connectors/中的教程使用AWS Glue连接器连接到BigQuery，但在执行完所有步骤后，我得到了一个： : java.lang.IllegalArgumentException: A project ID is required for this service but could not be determined from the builder
浏览 13提问于2021-07-02得票数 1
我们在GCP中有一个composer环境，我们希望将租户项目中的airflow数据库导出到BigQuery。有办法做到这一点吗？
浏览 3提问于2020-10-10得票数 0
1回答
我知道已经有一个airflow函数可以将文件从云存储传递到Big Query，正如我所做的，我在脚本中使用GCP进行连接的方式与没有airflow时相同，我调用PythonOperator来调用我在脚本中配置的函数，以读取云存储并将文件中的数据插入到Big Query中，但我收到错误消息：“获得意外的关键字参数'dag'”
这似乎很容易解决，但我真的不知道这是什么意思，因为我将DAG属性放在PythonOperator中
import json
import decimal
import airflow
from airflow import DAG
from airflow.
浏览 3提问于2019-06-22得票数 5
回答已采纳
我们有一个按小时运行的气流数据集，任务更新和覆盖BigQuery中的日期分区表。
在对这些表的查询和/或模式进行调整之后，我们想要回填几天的现有分区，但是回填所有运行都是巨大的浪费，因为每小时运行一次只会覆盖相同的分区24次，然后再转到第二天。
我们可以使用airflow list_dag_runs列出所有运行并筛选出每天的最后一个运行，但是是否有一种方法可以每天只回填/清除这些最后的运行，而不需要每天重新运行24个实例呢？
airflow clear和airflow backfill命令具有指定开始日期和结束日期的选项，但不指定特定的实例时间，因此它们将导致每次日期重新运行24次，它们都将执
浏览 3提问于2020-06-10得票数 0
回答已采纳
我试图从云存储桶中选择一个JSON文件，并使用Airflow将其转储到BigQuery中，但是，我得到了以下错误：
Access Denied: BigQuery BigQuery: Missing required OAuth scope. Need BigQuery or Cloud Platform write scope。
这是我的密码：
from datetime import timedelta, datetime
import json
from airflow.operators.bash_operator import BashOperator
from airflow.co
浏览 5提问于2022-11-03得票数 0
在Google BigQuery中，我使用气流触发负载作业。源文件由多个NDJSON文件组成。
这里是气流操作者(我认为与此无关)。(背景))：
load = GoogleCloudStorageToBigQueryOperator(
task_id=f"load",
bigquery_conn_id="bigquery_default",
pool="bigquery_insert",
destination_project_dataset_table="<HIDDEN>",
浏览 8提问于2020-02-14得票数 0
我正在使用Airflow中的BigQueryCheckOperator来了解BQ表中是否存在数据，但Dag使用此ERROR - <HttpError 404 when requesting https://bigquery.googleapis.com/bigquery/v2/projects/失败
有人能告诉我如何解决这个问题吗？
浏览 0提问于2020-07-30得票数 0
我们希望在Dag中的Dag触发器期间从UI读取cli输入传递给dag。我尝试了下面的代码，但它不起作用。在这里，我将输入传递为{“kpi”：“ID123”}，并希望在函数get_data_from_bq中打印此ip值。
from airflow import DAG
from airflow.utils.dates import days_ago
from airflow.operators.python_operator import PythonOperator
from airflow import models
from airflow.models import Variabl
浏览 1提问于2021-12-30得票数 0
我正在尝试将一串BigQuery SQL命令链接到一个ETL管道中，其中一些输出和输入将被加盖时间戳。
from datetime import timedelta
import airflow
from airflow import DAG
from airflow.contrib.operators.bigquery_operator import BigQueryOperator
DAG_NAME = 'foo'
default_args = {
'owner': 'airflow',
'depends_on_p
浏览 1提问于2018-07-12得票数 2
回答已采纳
1回答
我试图在google上的空气流编写器中执行以下DAG，并且一直得到相同的错误:未定义conn_id hard_coded_project_name
也许有人能帮我找到正确的方向？
from airflow.models import DAG
import os
from airflow.operators.dummy import DummyOperator
from airflow.operators.bash_operator import BashOperator
from airflow.operators.python_operator import PythonOperator
浏览 2提问于2021-11-29得票数 0
回答已采纳
从气流中读取BigQuery表并将结果保存为数据的最佳方法是什么？我有一个python模块来读取表，并且可以在本地运行这个模块来获取表：
def reading(ds, **kwargs):
project_id = '{project-id}'
credentials = service_account.Credentials.from_service_account_file('/usr/local/airflow/extras/credentials.json')
bqclient = bq.Client(credentials=
浏览 10提问于2021-09-22得票数 1
回答已采纳
我在airflow代码中使用了bigquery钩子。 查询示例：select count(*) from 'table-name';，因此它只返回1个整数作为结果。 如何将其保存在一个整型python变量中，而不是整个pandas数据帧中？ 下面是我的代码示例， hook = BigQueryHook(bigquery_conn_id=BQ_CON, use_legacy_sql=False)
bq_client = bigquery.Client(project = hook._get_field("project"), credentials = hook
浏览 53提问于2021-04-07得票数 1
我正在尝试从云存储中读取数据，并将数据写入BigQuery表。使用Pandas库读取GCS中的数据，并使用client.load_table_from_dataframe方法写入数据。我在Google cloud composer中以python操作符的身份执行此代码。当我执行代码时，出现了下面的错误。 [2020-06-23 17:09:36,119] {taskinstance.py:1059} ERROR - 'str' object has no attribute 'to_api_repr'@-@{"workflow": "D
浏览 14提问于2020-06-24得票数 0
回答已采纳
我有一个python函数，它在BigQuery中运行一个查询，并根据结果创建一个列表，然后将这个列表作为XCOM推送。
sql查询位于一个文件中，我希望使用参数将其传递给该sql。
我希望能够执行类似{{params.name}}允许的BigQueryOperator但使用PythonOperator的操作。
DAG成功运行，但它没有将DAG_NAME变量插入到sql中。
它只是将“{DAG_NAME}”称为静态字符串。
请帮帮忙，坚持一会儿。
我的密码
from airflow.models import (DAG, Variable)
import os
from airflow.ope
浏览 12提问于2022-01-14得票数 0
我正在试验数据管道的气流。不幸的是，到目前为止，我还不能让它与bigquery操作符一起工作。我已经尽了最大的努力寻找解决方案，但我还是被卡住了。我使用的是本地运行的顺序执行器。
下面是我的代码：
from airflow import DAG
from airflow.contrib.operators.bigquery_operator import BigQueryOperator
from datetime import datetime, timedelta
default_args = {
'owner': 'airflow',
&
浏览 3提问于2016-08-27得票数 10
回答已采纳
我试图在Composer中创建一个DAG。导入时，我将得到以下错误：
损坏的DAG: /home/airflow/gcs/dags/airflow_bigquery_v12.py无法导入名称_parse_data
这是DAG文件。如您所见，它试图将云存储文件复制到bigquery中：
import datetime
from datetime import timedelta, datetime
from airflow import DAG
from airflow.contrib.operators.bigquery_operator import BigQueryOperator
fro
浏览 0提问于2019-04-30得票数 2
回答已采纳
1回答
我对地理数据和气流都很陌生，所以如果我的问题不清楚，请原谅我，并要求精确。
我试图通过气流运行DAG (google composer)，从特定数据集中的表中读取数据，将特定列转换为地理类型，并将结果转储到另一个表中：
PROJECT = os.getenv("GCP_PROJECT")
default_args = {
"owner": "Airflow",
"depends_on_past": False,
"start_date": datetime(2020, 4, 1),
浏览 7提问于2020-04-13得票数 0
回答已采纳
我正在为分析团队开发数据目录，希望为他们提供气流作业信息的视图，包括dag id，执行，任务状态等存储到Bigquery表中，并暴露为仪表板视图。
任何建议，我应该如何处理design.Should，我使用Airflow CLI或Bashoperator来创建文件并写入Bigquery。
谢谢
浏览 1提问于2021-06-01得票数 1
我需要创建一个Airflow作业，将BigQuery表中的分区导出到给定范围的_PARTITIONDATE之间的GCS。我需要分区是在单独的文件与日期的分区。我如何才能做到这一点？
我已经尝试过使用airflow任务来使用SQL来获取_PARTITIONDATE，但是我可以通过编程来实现吗？
浏览 0提问于2019-09-17得票数 1
我希望使用App将BigQuery中的Google内容保存为一个表。
到目前为止，我们在BigQuery中创建了一个表"A“，它与Google同步。然后我们创建一个App，它查询表A并将其保存为表B。这个脚本非常类似于中描述的脚本。
这个解决方案工作得很好，但是我们必须创建一个新的表A，每次我们在google中更改表结构时。因此，我想知道如何将谷歌表的内容直接保存为表B，而不需要BigQuery中的表A。查询应使用第一行作为列名，并使用以下所有行作为内容。
浏览 3提问于2021-10-13得票数 1
我们已经为2.11版本编写了一个光束管道，但是当我们尝试使用DataflowOperator在Cloud Composer上运行它时，它使用的是SDK2.5版本。 是否有任何地方可以指定应该使用2.11？ 管道： import argparse
import apache_beam as beam
from apache_beam.io.gcp import gcsio
from apache_beam.options.pipeline_options import PipelineOptions
import logging
from google.cloud import storage
浏览 20提问于2019-04-18得票数 1
我试图为bigquery使用异步运算符；但是，
from airflow.providers.google.cloud.operators.bigquery import BigQueryCheckAsyncOperator
给出错误：
ImportError: cannot import name 'BigQueryCheckOperatorAsync' from 'airflow.providers.google.cloud.operators.bigquery'
中的文档提到了BigQueryCheckAsyncOperator的存在。
我用的是空气流量2
浏览 14提问于2022-11-18得票数 0
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
腾讯会议
实时音视频活动推荐
腾讯云开发者
