如何在BigQuery读写文件及通过DataFlow/Terraform导入CSV至BigQuery

BigQuery支持从本地文件、Cloud Storage（GCS）等源加载数据，以下是几种常用实现方式：

方法1：GCP控制台直接加载 #

• 进入BigQuery控制台，选择目标数据集，点击「创建表格」
• 在「数据源」中选择对应源：
  • 本地文件：上传CSV/JSON等格式文件，指定分隔符、表头设置
  • Cloud Storage：输入GCS文件路径（如gs://your-bucket/file.csv）
• 配置表格详情（表名、schema自动检测或手动定义），点击「创建表格」完成加载

方法2：使用bq命令行工具 #

• 加载本地文件到BigQuery：

bq load --source_format=CSV --skip_leading_rows=1 your-project:your-dataset.your-table ./local-file.csv

• 加载GCS文件到BigQuery：

bq load --source_format=CSV --skip_leading_rows=1 your-project:your-dataset.your-table gs://your-bucket/file.csv

• 参数说明：--skip_leading_rows=1 用于跳过表头，--source_format 指定文件格式（支持CSV、JSON、AVRO等）

方法3：Python客户端库加载 #

先安装依赖：

pip install google-cloud-bigquery

示例代码（从GCS加载CSV到BigQuery）：

from google.cloud import bigquery

client = bigquery.Client()
dataset_id = "your-project.your-dataset"
table_id = f"{dataset_id}.your-table"

job_config = bigquery.LoadJobConfig(
    source_format=bigquery.SourceFormat.CSV,
    skip_leading_rows=1,
    autodetect=True,  # 自动检测schema
)

uri = "gs://your-bucket/file.csv"

load_job = client.load_table_from_uri(
    uri, table_id, job_config=job_config
)

load_job.result()  # 等待加载完成

destination_table = client.get_table(table_id)
print(f"Loaded {destination_table.num_rows} rows.")

以下是自定义Python代码（非模板）实现DataFlow作业的步骤：

步骤1：准备环境 #

安装Apache Beam及GCP依赖：

pip install apache-beam[gcp]

步骤2：编写DataFlow Python代码 #

创建csv_to_bigquery.py文件，内容如下（可根据自身CSV结构调整解析逻辑）：

import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions, StandardOptions

class ParseCSV(beam.DoFn):
    def process(self, element):
        # 按逗号分割CSV行，匹配你的字段数量与类型
        columns = element.split(',')
        return [{
            'id': int(columns[0]),
            'name': columns[1],
            'value': float(columns[2])
        }]

def run():
    pipeline_options = PipelineOptions()
    # 配置DataFlow运行参数
    pipeline_options.view_as(StandardOptions).runner = 'DataflowRunner'
    pipeline_options.view_as(StandardOptions).project = 'your-project-id'
    pipeline_options.view_as(StandardOptions).region = 'us-central1'
    pipeline_options.view_as(StandardOptions).temp_location = 'gs://your-bucket/temp'
    pipeline_options.view_as(StandardOptions).staging_location = 'gs://your-bucket/staging'

    with beam.Pipeline(options=pipeline_options) as p:
        # 从GCS读取CSV，跳过表头
        lines = p | 'Read CSV from GCS' >> beam.io.ReadFromText('gs://your-bucket/input-data.csv', skip_header_lines=1)
        
        # 解析CSV行成结构化数据
        parsed_data = lines | 'Parse CSV' >> beam.ParDo(ParseCSV())
        
        # 写入BigQuery
        parsed_data | 'Write to BigQuery' >> beam.io.WriteToBigQuery(
            table='your-project-id:your-dataset.your-table',
            schema='id:INTEGER,name:STRING,value:FLOAT',
            write_disposition=beam.io.BigQueryDisposition.WRITE_TRUNCATE,
            create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED
        )

if __name__ == '__main__':
    run()

步骤3：在GCP控制台提交DataFlow作业 #

1. 打开DataFlow控制台，点击「创建作业」
2. 输入作业名称，选择对应区域
3. 在「Python文件路径」中上传本地的csv_to_bigquery.py，或输入已上传的GCS路径（如gs://your-bucket/csv_to_bigquery.py）
4. 按需添加额外参数，点击「运行作业」等待执行完成

以下是Terraform配置示例，替代控制台操作实现资源创建与DataFlow作业提交：

步骤1：编写Terraform配置文件（main.tf） #

provider "google" {
  project = "your-project-id"
  region  = "us-central1"
}

# 可选：创建GCS存储桶用于存放代码、输入数据及临时文件
resource "google_storage_bucket" "dataflow_bucket" {
  name          = "your-unique-bucket-name"
  location      = "US"
  storage_class = "STANDARD"
}

# 可选：创建目标BigQuery表（若表未预先存在）
resource "google_bigquery_table" "target_table" {
  dataset_id = "your-dataset"
  table_id   = "your-table"

  schema = <<EOF
[
  {"name": "id", "type": "INTEGER"},
  {"name": "name", "type": "STRING"},
  {"name": "value", "type": "FLOAT"}
]
EOF
}

# 提交DataFlow作业（需先将csv_to_bigquery.py上传至GCS桶）
resource "google_dataflow_job" "csv_to_bq" {
  name        = "csv-to-bigquery-job"
  region      = "us-central1"
  python_file = "gs://${google_storage_bucket.dataflow_bucket.name}/csv_to_bigquery.py"
  
  parameters = {
    input = "gs://${google_storage_bucket.dataflow_bucket.name}/input-data.csv"
    output = "${google_bigquery_table.target_table.project}:${google_bigquery_table.target_table.dataset_id}.${google_bigquery_table.target_table.table_id}"
  }

  temp_gcs_location = "gs://${google_storage_bucket.dataflow_bucket.name}/temp"
}

步骤2：执行Terraform命令 #

• 初始化Terraform：

terraform init

• 预览执行计划：

terraform plan

• 应用配置（创建资源并提交作业）：

terraform apply

说明：若已存在GCS存储桶和BigQuery表，可移除对应资源块，仅保留DataFlow作业配置部分。

内容的提问来源于stack exchange，提问作者TEJAS