当前位置：首页 > news >正文

pandas数据存到informix数据库

news 来源：原创 2025/5/17 12:27:53

紧接上文python 连接infomix，结合pandas，补充csdn在这方面的经验
。

由于无法通过sqlalchemy连接数据库ibm的informix数据库。得用jaydebeapi的jar包。

那么这篇文章就是介绍如何将十几万条的pandas的数据存到informix中。

ok，首先我们读取csv文件

batch_df=pd.read_csv(filepath,encoding="ANSI",thousands=",",chunksize=1000)
for idx,df in enumerate(batch_df):
	# 增加插入时间，也可以放在下载里面处理datetime带有时间戳，所以要格式化一下
	df=df.assign(下载日期=datetime.datetime.today().strftime("%Y-%m-%d %H:%M:%S"))
    dfData = filter_dfData(df) #按照自己的要求进行过滤
    sql, insert_statements = generate_insert_sql(dfData, tablename) #生产sql和参数
    sql_to_db(sql, insert_statements, conn)

在数据过滤中，值得注意的是pandas中null是以np.nan填充的，而非None。

def filter_dfData(df):#数据过滤
	# xxx_str是形如 （city VARCHAR(50), -- 地市机构代码）的sql建表语句，
	data_dict=xxx_str.split("\n") 
	my_dict = {itemlist[-1]: itemlist[0] for itemlist in (item.split(" ") for item in data_dict)}
	df=df.rename(columns=my_dict)# 按照自己的字典替换df的名字
    df.replace(np.nan, None, inplace=True)#将df的空值替换为None。

然后自定义我们需要的sql语句，可喜可贺的是，jaydebeapi支持statement和many的方法，这样1、可以防止sql注入
2、sql和数据分开了
大大提升了效率。

# 接受一个df和tabl_name，返回sql和数据
def generate_insert_sql(df, table_name):
    insert_statements = []
    columns = ', '.join(df.columns)  # 生成字段名
    placeholders = ", ".join(["?"] * len(df.columns))  # 生成占位符
    sql = f"INSERT INTO {table_name} ({columns}) VALUES ({placeholders});"  # 生成sql语句
    for index, row in df.iterrows():
        values = []
        for value in row:  # 将float和str分开，有单引号的区别
            if isinstance(value, (float)):
                # 数值类型直接使用
                processed = value
            elif value == None:
                processed = value
            else:
                processed = str(value)
            values.append(processed)
        values_str = tuple(values)  # 每条数据转为元组
        insert_statements.append(values_str)  # 组合多条数据
    return sql, insert_statements

最后将代码交由conn执行

def sql_to_db(sql, insert_statements, conn):
    try:
        cursor = conn.cursor()
        cursor.executemany(sql, insert_statements)
        print("提交成功")
    except Exception as e:
        print("提交失败", e)
    finally:
        cursor.close()