И это мой код ниже.
Я получаю приведенную ниже ошибку, когда запускаю приведенный ниже код.Я пытаюсь преобразовать файлы gvcf / vcf в моем облачном хранилище Google в bigquery с помощью API gcp-variable-transforms.
[2018-06-06 16: 46: 42,589] {models.py:1428} INFO - Выполнение 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 / GcsToBigQu.py '] [2018-06-06 16: 46: 43,204] {base_task_runner.py:98} INFO - подзадача: [2018-06-06 16: 46: 43,202] { init .py: 45} INFO - Использование executor 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 из /apps/airflow/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} INFO - подзадача: [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)