DataFlow 使用 Airflow DataflowHook.start_python_dataflow 失败并返回代码 1
Posted
技术标签:
【中文标题】DataFlow 使用 Airflow DataflowHook.start_python_dataflow 失败并返回代码 1【英文标题】:DataFlow failed with return code 1 with Airflow DataflowHook.start_python_dataflow 【发布时间】:2018-06-06 22:10:26 【问题描述】:这是我下面的代码。
当我运行以下代码时,出现以下错误。我正在尝试使用 gcp-variant-transforms api 将我的谷歌云存储中的 gvcf/vcf 文件转换为 bigquery。
[2018-06-06 16:46:42,589] models.py:1428 信息 - 执行于 2018-06-06 21:46:34.252526 [2018-06-06 16:46:42,589] base_task_runner.py:115 信息 - 运行:['bash', '-c', u'airflow run GcsToBigQuery gcsToBigquery_ID 2018-06-06T21:46:34.252526 -- job_id 168 --raw -sd DAGS_FOLDER/GcsToBigQuery.py'] [2018-06-06 16:46:43,204] base_task_runner.py:98 信息 - 子任务:[2018-06-06 16:46:43,202] init.py:45 信息- 使用执行器 SequentialExecutor [2018-06-06 16:46:43,284] base_task_runner.py:98 INFO - 子任务:[2018-06-06 16:46:43,283] models.py:189 INFO - 从 / 填充 DagBag应用程序/气流/dags/GcsToBigQuery.py [2018-06-06 16:46:43,853] base_task_runner.py:98 INFO - 子任务:[2018-06-06 16:46:43,852] gcp_dataflow_hook.py:111 INFO - 开始等待 DataFlow 进程完全的。 [2018-06-06 16:46:46,931] base_task_runner.py:98 信息 - 子任务:[2018-06-06 16:46:46,930] GcsToBigQuery.py:48 错误 - 状态:失败:gcsToBigquery:无法运行:DataFlow 失败,返回码 1 [2018-06-06 16:46:46,931] base_task_runner.py:98 信息 - 子任务:[2018-06-06 16:46:46,930] python_operator.py:90 信息 - 完成。返回值为:无
请帮我解决这个问题。谢谢!
from datetime import datetime, timedelta
from airflow import DAG
from airflow.contrib.hooks.gcp_dataflow_hook import DataFlowHook
from airflow.operators.python_operator import PythonOperator
import logging
default_args =
'owner': 'My Name',
'depends_on_past': False,
'start_date': datetime(2018, 6, 6),
'email': ['MY Email'],
'email_on_failure': True,
'email_on_retry': False,
'retries': 2,
'retry_delay': timedelta(minutes=5)
dag = DAG('GcsToBigQuery', default_args=default_args,
description='To move GVCF/VCF files from Google Cloud Storage to Big Query',
schedule_interval='@once',
start_date=datetime(2018, 6, 6))
dataflow_py_file = 'gcp_variant_transforms.vcf_to_bq'
PY_OPTIONS = ['-m']
DATAFLOW_OPTIONS_PY =
"project": "project-Name",
"input_pattern": "gs://test-gvcf/1000-genomes.vcf",
"output_table": "trc-mayo-projectsample:1000genomicsID.1000_genomesSamp",
"staging_location": "gs://test-gvcf/vcftobq/staging",
"temp_location": "gs://test-gvcf/vcftobq/temp",
"job_name": "dataflowstarter25",
#"setup_file": "./setup.py",
"runner": "DataflowRunner"
def gcsToBigquery():
try:
dataflowHook = DataFlowHook(gcp_conn_id='google_cloud_platform_id')
dataflowHook.start_python_dataflow(task_id='dataflowStarter2_ID', variables=DATAFLOW_OPTIONS_PY,
dataflow=dataflow_py_file, py_options=PY_OPTIONS)
except Exception as e:
logging.error("Status : FAIL : gcsToBigquery: Not able to run: " + str(e.message))
gcsToBigquery_task = PythonOperator(task_id='gcsToBigquery_ID',
python_callable=gcsToBigquery,
dag=dag)
【问题讨论】:
【参考方案1】:可以通过使用DataflowPythonOperator 并在云实例中安装gcp_variant_transforms
API 来避免此问题。
安装所需组件的命令:
sudo pip install git+https://github.com/googlegenomics/gcp-variant-transforms.git
如果有人也遇到此问题,您可以查看此post,其中详细说明了srikanth 成功执行代码所遵循的步骤。
【讨论】:
以上是关于DataFlow 使用 Airflow DataflowHook.start_python_dataflow 失败并返回代码 1的主要内容,如果未能解决你的问题,请参考以下文章
将 BigQuery 的聚合具体化转储到 SQL 服务器、Dataflow 与 Airflow
Apache Airflow 1.9:作业结束时的数据流异常