气流 bigquery_to_gcs 运营商改变 field_delimiter答案

【问题标题】：Airflow bigquery_to_gcs operator changing field_delimiter气流 bigquery_to_gcs 运营商改变 field_delimiter
【发布时间】：2023-01-23 23:21:40
【问题描述】：

我正在尝试使用 Airflow 运算符 BigQueryToGCSOperator & forcing字段分隔符是管道 (|) ，但是文件的输出总是以逗号 (,) 分隔。

我也尝试过具有相同行为的运算符 BigQueryToCloudStorageOperator。

知道我在这里做错了什么吗？

from airflow.providers.google.cloud.transfers.bigquery_to_gcs import (
    BigQueryToGCSOperator,
  
data_to_gcs = BigQueryToGCSOperator(
        task_id="BigQuery_to_GoogleCloudBucket",
        gcp_conn_id="google_cloud_default",
        project_id=project_id,
        source_project_dataset_table=f"{project_id}.{temp_dataset_id}.{temp_table}",
        location="EU",
        print_header=True,
        destination_cloud_storage_uris=destination_uri,
        export_format="csv",
        field_delimiter="|",
    )

预先感谢您的回复。

【问题讨论】：

标签： google-cloud-platform airflow google-cloud-composer

【解决方案1】：

通常，如果您将 export_format 字段设置为 CSV（大写而不是小写）和 field_delimiter，它应该可以工作：

from airflow.providers.google.cloud.transfers.bigquery_to_gcs import (
    BigQueryToGCSOperator,
  
data_to_gcs = BigQueryToGCSOperator(
        task_id="BigQuery_to_GoogleCloudBucket",
        gcp_conn_id="google_cloud_default",
        project_id=project_id,
        source_project_dataset_table=f"{project_id}.{temp_dataset_id}.{temp_table}",
        location="EU",
        print_header=True,
        destination_cloud_storage_uris=destination_uri,
        export_format="CSV",
        field_delimiter="|",
    )

我在 Airflow 代码中看到了这段代码 sn-p，我认为如果将 export_format 和 csv 设置为小写值，可能会导致问题：

if self.export_format == 'CSV':
     # Only set fieldDelimiter and printHeader fields if using CSV.
     # Google does not like it if you set these fields for other export
     # formats.
     configuration['extract']['fieldDelimiter'] = self.field_delimiter
     configuration['extract']['printHeader'] = self.print_header

在您的情况下，不会调用此代码 sn-p 并且运算符采用 field_delimiter 的默认值，即 ,

在这里，您可以在 Airflow 代码中看到此运算符的构造函数中使用的默认值：

def __init__(
    self,
    *,
    source_project_dataset_table: str,
    destination_cloud_storage_uris: List[str],
    compression: str = 'NONE',
    export_format: str = 'CSV',
    field_delimiter: str = ',',
    print_header: bool = True,
    gcp_conn_id: str = 'google_cloud_default',
    bigquery_conn_id: Optional[str] = None,
    delegate_to: Optional[str] = None,
    labels: Optional[Dict] = None,
    location: Optional[str] = None,
    impersonation_chain: Optional[Union[str, Sequence[str]]] = None,
    **kwargs,
)

【讨论】：