如何将指定URL的csv.gz文件导入Google BigQuery及路径选择咨询

一、批量同步数据到GCS：不用逐个上传 #

不用手动逐个上传，有两种高效方法：

• 用gsutil命令并行拉取：直接从NOAA的HTTP源批量复制到GCS，加-m参数开启多线程加速，命令如下：

  gsutil -m cp "https://www1.ncdc.noaa.gov/pub/data/ghcn/daily/by_year/*.csv.gz" gs://your-bucket-name/ghcn-daily/
  

• 用Cloud Storage Transfer Service：适合超大规模文件量，在GCP控制台创建传输任务，设置HTTP源为NOAA的文件目录，自动批量拉取并同步到GCS，无需本地执行脚本，稳定性更高。

二、先导原始数据还是先清洗？看清洗复杂度选 #

优先选：先导原始数据到BigQuery，再在BQ内清洗 #

如果你的清洗逻辑是字段转换、过滤无效值、单位换算这类简单操作，直接在BigQuery里处理效率更高、成本更低：

1. 先创建指向GCS原始文件的外部表（GHCN的csv没有表头，要手动定义Schema）：

   CREATE OR REPLACE EXTERNAL TABLE `your-project.your-dataset.ghcn_raw` (
     id STRING,
     date DATE,
     element STRING,
     data_value INT64,
     mflag STRING,
     qflag STRING,
     sflag STRING,
     obs_time STRING
   )
   OPTIONS (
     format = 'CSV',
     uris = ['gs://your-bucket-name/ghcn-daily/*.csv.gz'],
     skip_leading_rows = 0,
     field_delimiter = ','
   );
   

2. 写SQL完成清洗，生成最终的内部表：

   CREATE OR REPLACE TABLE `your-project.your-dataset.ghcn_cleaned` AS
   SELECT
     id,
     date,
     element,
     data_value / 10 AS data_value, -- 原始值是放大10倍的整数，转换为实际数值
     mflag,
     qflag,
     sflag,
     obs_time
   FROM `your-project.your-dataset.ghcn_raw`
   WHERE qflag IS NULL; -- 过滤有质量标记的无效数据
   

仅当清洗逻辑复杂时，先清洗再导入 #

如果需要跨文件关联、自定义复杂逻辑（BQ的UDF满足不了），再用PySpark处理：

1. 先把数据同步到GCS；
2. 配置PySpark的GCS和BigQuery连接器，读取GCS上的文件清洗后直接写入BQ：

   from pyspark.sql import SparkSession
   
   spark = SparkSession.builder \
     .appName("GHCNDataCleaning") \
     .config("spark.jars.packages", "com.google.cloud.spark:spark-bigquery-with-dependencies_2.12:0.35.1") \
     .getOrCreate()
   
   # 读取GCS上的原始压缩文件
   raw_df = spark.read.csv(
     "gs://your-bucket-name/ghcn-daily/*.csv.gz",
     schema="id STRING, date DATE, element STRING, data_value INT64, mflag STRING, qflag STRING, sflag STRING, obs_time STRING"
   )
   
   # 自定义清洗逻辑示例
   cleaned_df = raw_df.filter(raw_df.qflag.isNull()) \
                      .withColumn("data_value", raw_df.data_value / 10)
   
   # 直接写入BigQuery
   cleaned_df.write.format("bigquery") \
     .option("table", "your-project.your-dataset.ghcn_cleaned") \
     .option("temporaryGcsBucket", "your-bucket-name/temp") \
     .mode("overwrite") \
     .save()
   

三、整体最优流程 #

1. 用Cloud Storage Transfer Service或gsutil -m cp批量同步NOAA的csv.gz到GCS；
2. 若清洗逻辑简单，直接在BigQuery创建外部表+SQL清洗生成内部表；
3. 若清洗逻辑复杂，用PySpark读取GCS数据清洗后写入BigQuery。

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