引入整行数据（通过气流）时，Google GCS 到 BIGQUERY 失败

Question

我使用 GoogleCloudStorageToBigQueryOperator 的目的是将一系列类似结构的文件带入大查询中。

因为我知道由于摄取的限制，需要在 bigquery 中解析文件，所以我想将每一行数据作为单个数据元素（一个字段）引入。以下是我的代码：

GCS_to_GBQ_Raw = GoogleCloudStorageToBigQueryOperator(
    task_id='GCS_to_GBQ_Raw',
    bucket='files',
    source_objects=['To_Process/*.csv'],
    source_format='CSV',
    destination_project_dataset_table='DS.RAW',
    schema_fields=[
       {'name': 'datarow', 'type': 'STRING', 'mode': 'NULLABLE'}
    ],
    field_delimiter='\t',
    autodetect=False,
    skip_leading_rows=1,
    write_disposition='WRITE_TRUNCATE',
    quote_character='µ',
    google_cloud_storage_conn_id='GCP_Staging',
    bigquery_conn_id='GCP_Staging',
    dag=dag)

我尝试了以下方法：

1. field_delimiter - 我尝试过模糊的 ascii 字符，例如节字符、管道和制表符。我知道文件中的数据没有这些。
2. autodetect=真假。
3. quote_character=''、'""'、当前字符 mu，以及我知道的其他字符不在文件中。

我每次运行时都会收到一组随机错误：

• 列太多
• 引用和字段结尾之间的数据
• 正在尝试追加

我该如何进行这项工作？如何引入完整的行，以便在 bq 中解析它们？

谢谢！

Answer 1

我不确定您的文件看起来如何，如果您可以提供一些示例记录，那么我们可以提供更好的帮助。 如果我正确理解了您的问题，那么下面的代码应该可以工作：

import datetime as dt
from airflow.models import DAG
from airflow.contrib.operators.gcs_to_bq import GoogleCloudStorageToBigQueryOperator
# from airflow.utils.dates import days_ago
lasthour = dt.datetime.utcnow() - dt.timedelta(hours=1)

args = {
 'owner': 'airflow',
 'start_date': lasthour,
 'depends_on_past': False,
 'job_name': 'employee-test',
}
dag = DAG(
 dag_id='just-check',
 schedule_interval=None,
 default_args=args
)

load_csv = GoogleCloudStorageToBigQueryOperator(
    task_id='gcs_to_bq_example',
    bucket='testing-bucket',
    source_objects=['employee/*.csv'],
    field_delimiter='|',
    skip_leading_rows=1,
    autodetect=False,
    destination_project_dataset_table='project_id.raw.gcs_to_bq_table',
    schema_fields=[
       {'name': 'datarow', 'type': 'STRING', 'mode': 'NULLABLE'}
    ],
    write_disposition='WRITE_TRUNCATE',
    dag=dag)

运行此 DAG 后，我的 BQ 表的每一行数据都有一列。