Ошибка авторизации при запуске воздушного потока через облачный композитор

Question

Я получаю сообщение об ошибке при попытке запустить DAG из облачного композитора с помощью GoogleCloudStorageToBigQueryOperator.

Последняя ошибка: {'причина': 'недействительно', 'местоположение': 'gs: // xxxxxx / xxxx.csv ', и когда я перехожу по URL-ссылке на ошибку ...

{
  "error": {
    "code": 401,
    "message": "Request is missing required authentication credential. Expected OAuth 2 access token, login cookie     or other valid authentication credential. See https://developers.google.com/identity/sign-in/web/devconsole-    project.",
    "errors": [
      {
        "message": "Login Required.",
        "domain": "global",
        "reason": "required",
        "location": "Authorization",
        "locationType": "header"
      }
    ],
    "status": "UNAUTHENTICATED"
  }
}

Я настроил соединение с облачным хранилищем ...

Conn Id My_Cloud_Storage

Conn Type Google Cloud Platform

Идентификатор проекта xxxxxx

Путь к ключевому файлу /home/airflow/gcs/data/xxx.json

Ключевой файл JSON

Области(через запятую) https://www.googleapis.com/auth/cloud-platform

Код ..

from __future__ import print_function

import datetime

from airflow import models
from airflow import DAG
from airflow.operators import bash_operator
from airflow.operators import python_operator
from airflow.contrib.operators.gcs_to_bq import GoogleCloudStorageToBigQueryOperator

default_dag_args = {
    # The start_date describes when a DAG is valid / can be run. Set this to a
    # fixed point in time rather than dynamically, since it is evaluated every
    # time a DAG is parsed. See:
    # https://airflow.apache.org/faq.html#what-s-the-deal-with-start-date
    'start_date': datetime.datetime(2019, 4, 15),
}
with models.DAG(
        'Ian_gcs_to_BQ_Test',
        schedule_interval=datetime.timedelta(days=1),
        default_args=default_dag_args) as dag:

    load_csv = GoogleCloudStorageToBigQueryOperator(
        task_id='gcs_to_bq_test',
        bucket='xxxxx',
        source_objects=['xxxx.csv'],
        destination_project_dataset_table='xxxx.xxxx.xxxx',
        google_cloud_storage_conn_id='My_Cloud_Storage',
        schema_fields=[
            {'name':'AAAA','type':'INTEGER','mode':'NULLABLE'},
            {'name':'BBB_NUMBER','type':'INTEGER','mode':'NULLABLE'},   
        ],
        write_disposition='WRITE_TRUNCATE',
        dag=dag)

Answer 1

Хорошо, теперь это исправлено. Оказывается, он не работал из-за строки заголовка в файле, как только я убрал, что он работал нормально. Довольно раздражающие, вводящие в заблуждение сообщения об ошибках о неверных местоположениях и авторизации.