Snowflake数据加载异常：DataFrame写入时行被识别为列名

问题分析 #

核心问题有两点：一是CSV文件会因空值出现None字符串，二是DataFrame错误将每行数据识别为列名。后者大概率是数据结构处理不当，导致pd.DataFrame()构造时行/列识别异常，或是空值处理过程破坏了行数据的结构。

解决方案 #

以下是修正后的代码，同时解决空值转None和行结构错误的问题：

import os
import pandas as pd
from datetime import datetime
from airflow.providers.snowflake.hooks.snowflake import SnowflakeHook

def execute(self, context):
    source_hook = SnowflakeHook(snowflake_conn_id=self.source_conn_id)
    dest_hook = SnowflakeHook(snowflake_conn_id=self.dest_conn_id)

    # 从源Snowflake获取数据
    rows = source_hook.get_records(self.sql)

    # 定义临时文件路径
    directory = '/opt/airflow/logs/'
    filename = 'abc_test.txt'
    file_path = os.path.join(directory, filename)

    # 处理日期格式和空值：将日期转成Snowflake兼容格式，空值替换为空字符串
    processed_rows = []
    for row in rows:
        row_list = list(row)
        for idx, value in enumerate(row_list):
            if isinstance(value, datetime):
                row_list[idx] = value.strftime('%Y-%m-%d %H:%M:%S')
            elif value is None:
                # 替换空值为空字符串，避免CSV出现'None'
                row_list[idx] = ''
        processed_rows.append(row_list)

    # 写入CSV文件，确保结构正确
    with open(file_path, 'w+') as temp_file:
        df = pd.DataFrame(processed_rows)
        print(df)
        # header=False 确保不写入列名，sep='|' 指定分隔符，na_rep='' 兜底处理NaN值
        df.to_csv(temp_file, index=False, header=False, sep='|', na_rep='')

关键修正点 #

• 空值双重处理：新增elif value is None分支替换空值为空字符串，同时to_csv中添加na_rep=''，彻底避免None出现在文件中。
• 数据结构维护：改用processed_rows列表收集处理后的行，避免直接修改原rows时可能出现的索引错误，确保每行都是完整的列表结构，让pd.DataFrame()正确识别行和列。
• 参数明确化：显式指定na_rep=''，覆盖DataFrame中NaN值的默认输出行为。

额外检查项 #

1. 确认source_hook.get_records(self.sql)返回的是二维列表（每个元素是一行数据的元组/列表），如果返回一维结构会导致DataFrame构造异常。
2. 若目标Snowflake表有固定列顺序，建议创建DataFrame时指定列名（如df = pd.DataFrame(processed_rows, columns=['col1', 'col2', ...])），避免列顺序混乱。

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