基于Red Hat Linux用Snowflake ODBC Driver批量加载多结构CSV至对应表

针对Red Hat Linux上数百个100MB级、列数各异的CSV文件，推荐使用Snowflake官方的COPY INTO命令结合Shell脚本实现高效批量加载，该方式支持自动匹配列名，无需手动指定字段；若需基于已安装的ODBC驱动实现，也可通过Python脚本动态处理表头与数据。

方案一：SnowSQL + Shell脚本（高效推荐） #

SnowSQL是Snowflake的官方CLI工具，结合COPY INTO的列名匹配特性，能完美适配列数各异的场景，且加载效率远高于ODBC单条插入。

步骤1：安装并配置SnowSQL #

1. 下载Red Hat兼容的SnowSQL RPM包并安装：

   sudo rpm -i snowsql-<version>-linux_x86_64.rpm
   

2. 配置连接参数（也可直接在脚本中指定）：

   snowsql -c my_connection -a <your_account> -u <your_user> -p <your_password> -w <your_warehouse> -d <your_db> -s <your_schema>
   

步骤2：创建通用CSV文件格式 #

创建支持列名匹配的文件格式，确保CSV表头与Snowflake表列自动对应：

CREATE OR REPLACE FILE_FORMAT my_csv_format
TYPE = CSV
FIELD_OPTIONALLY_ENCLOSED_BY = '"'
SKIP_HEADER = 1
MATCH_BY_COLUMN_NAME = CASE_INSENSITIVE;

步骤3：编写批量处理Shell脚本 #

假设CSV文件存放在/data/csv_files/目录，文件名与目标表名一致（如customer_data.csv对应customer_data表），脚本如下：

#!/bin/bash

# 配置Snowflake连接参数
ACCOUNT="<your_account>"
USER="<your_user>"
PASSWORD="<your_password>"
WAREHOUSE="<your_warehouse>"
DATABASE="<your_db>"
SCHEMA="<your_schema>"
STAGE_NAME="csv_upload_stage"
CSV_DIR="/data/csv_files/"

# 创建临时上传阶段（若不存在）
snowsql -a $ACCOUNT -u $USER -p $PASSWORD -w $WAREHOUSE -d $DATABASE -s $SCHEMA -q "CREATE OR REPLACE STAGE $STAGE_NAME FILE_FORMAT = my_csv_format;"

# 遍历所有CSV文件
for csv_file in "$CSV_DIR"*.csv; do
    # 提取表名（去掉路径和后缀）
    table_name=$(basename "$csv_file" .csv)
    
    # 自动创建表（若表不存在）
    snowsql -a $ACCOUNT -u $USER -p $PASSWORD -w $WAREHOUSE -d $DATABASE -s $SCHEMA -q "DESCRIBE TABLE $table_name" > /dev/null 2>&1
    if [ $? -ne 0 ]; then
        # 从CSV表头生成建表语句（默认列类型为STRING，可按需调整）
        headers=$(head -1 "$csv_file" | tr ',' '\n' | sed 's/^/"/g; s/$/" STRING/g' | tr '\n' ', ')
        headers=${headers%, }
        snowsql -a $ACCOUNT -u $USER -p $PASSWORD -w $WAREHOUSE -d $DATABASE -s $SCHEMA -q "CREATE TABLE $table_name ($headers);"
        echo "Created table $table_name"
    fi
    
    # 上传文件到Snowflake阶段
    snowsql -a $ACCOUNT -u $USER -p $PASSWORD -w $WAREHOUSE -d $DATABASE -s $SCHEMA -q "PUT file://$csv_file @$STAGE_NAME AUTO_COMPRESS = FALSE;"
    
    # 批量加载到目标表，自动匹配列名
    snowsql -a $ACCOUNT -u $USER -p $PASSWORD -w $WAREHOUSE -d $DATABASE -s $SCHEMA -q "COPY INTO $table_name FROM @$STAGE_NAME/$(basename $csv_file) FILE_FORMAT = my_csv_format MATCH_BY_COLUMN_NAME = CASE_INSENSITIVE;"
    
    # 清理阶段文件（可选）
    snowsql -a $ACCOUNT -u $USER -p $PASSWORD -w $WAREHOUSE -d $DATABASE -s $SCHEMA -q "REMOVE @$STAGE_NAME/$(basename $csv_file);"
    
    echo "Successfully loaded $csv_file into $table_name"
done

脚本使用说明 #

1. 替换参数中的<your_account>、<your_user>等为实际信息
2. 赋予脚本执行权限：chmod +x load_csv_to_snowflake.sh
3. 运行脚本：./load_csv_to_snowflake.sh

方案二：Python + ODBC（适配已有ODBC环境） #

若需基于已安装的Snowflake ODBC驱动实现，可通过Python动态读取CSV表头，生成对应INSERT语句批量插入：

步骤1：安装依赖 #

pip install pyodbc pandas

步骤2：编写Python脚本 #

import pyodbc
import pandas as pd
import os

# ODBC连接字符串
conn_str = (
    "DRIVER={SnowflakeDSIIDriver};"
    "ACCOUNT=<your_account>;"
    "USER=<your_user>;"
    "PASSWORD=<your_password>;"
    "WAREHOUSE=<your_warehouse>;"
    "DATABASE=<your_db>;"
    "SCHEMA=<your_schema>;"
)

csv_dir = "/data/csv_files/"

# 遍历CSV文件
for filename in os.listdir(csv_dir):
    if not filename.endswith(".csv"):
        continue
        
    file_path = os.path.join(csv_dir, filename)
    table_name = os.path.splitext(filename)[0]
    
    # 读取CSV表头，生成建表语句（若表不存在）
    with open(file_path, 'r') as f:
        headers = f.readline().strip().split(',')
    
    with pyodbc.connect(conn_str) as conn:
        cursor = conn.cursor()
        # 检查表是否存在
        cursor.execute(f"SELECT COUNT(*) FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_NAME = '{table_name}'")
        if cursor.fetchone()[0] == 0:
            # 生成建表语句，默认列类型为STRING
            columns_def = ", ".join([f'"{col}" STRING' for col in headers])
            cursor.execute(f"CREATE TABLE {table_name} ({columns_def})")
            conn.commit()
            print(f"Created table {table_name}")
        
        # 分块读取CSV并批量插入（避免内存溢出）
        for chunk in pd.read_csv(file_path, chunksize=1000):
            # 生成INSERT语句
            placeholders = ", ".join(["?" for _ in headers])
            insert_sql = f"INSERT INTO {table_name} ({', '.join([f'"{col}"' for col in headers])}) VALUES ({placeholders})"
            # 批量执行插入
            cursor.executemany(insert_sql, chunk.values.tolist())
            conn.commit()
    
    print(f"Loaded {file_path} into {table_name} successfully")

关键注意事项 #

• 确保CSV表头与Snowflake目标表的列名一致（大小写不敏感）
• 大文件场景优先选择方案一，COPY INTO是Snowflake优化的批量加载方式，性能远高于ODBC批量插入
• 若CSV存在特殊字符或格式问题，需调整文件格式的参数（如ESCAPE、FIELD_DELIMITER等）

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