处理大型 csv 文件。如何使用 python 在 MySQL 数据库中插入 csv 数据？答案

【问题标题】：Working with large csv files . How to insert csv data in MySQL database using python?处理大型 csv 文件。如何使用 python 在 MySQL 数据库中插入 csv 数据？
【发布时间】：2021-09-20 14:13:49
【问题描述】：

我对如何将例如 10 GB 的 .csv 文件插入 MySQL 数据库感兴趣。我使用 pandas 和 pyspark 读取 csv 文件，然后将 csv 标头添加到列表中（如果 csv 文件没有标头，我正在使用 spark 添加它）。然后我正在解析列表并替换 MySQL 插入代码的字符 -->

mydb.cursor().execute("CREATE TABLE " + table_name + " (id INT NOT NULL AUTO_INCREMENT," + column_names + ", PRIMARY KEY (id))")

然后我将没有标题行的整行添加到 list ，然后再次解析以将 'name' 替换为 `name` 。所以我正在编辑列表

query = "insert into `"+ table_name +"` (" + column_names + ") values (" + row_value + ")"

这对于小型 csv 文件非常有效。但是对于大文件，进程会因为内存不足而崩溃。

但是大的 csv 文件呢？是否有将大型 csv 文件数据插入 MySQL 的解决方法？或者您是否有处理大型 csv 文件以将它们插入 MySQL 而不会出现内存不足问题的代码示例？

我认为如果我将大的 csv 文件分成小文件然后插入，它会更好地用于内存。并且可能有一些更好的方法可以将这种大小的数据插入 MySQL。

谢谢。

【问题讨论】：

MySQL可以直接使用LOAD DATA INFILE读取CSV文件。为什么要通过 Python 来做？
但它需要先创建表对吗？所以我必须阅读 csv 文件才能看到列的名称。我想要上传数据的自动化方式，这就是我想用 Python 来做的原因。
Python 可以读取 CSV 的第一行以获取列名并创建表。然后使用LOAD DATA INFILE 将内容加载到表中。但是你会从哪里得到数据类型呢？
无论如何，pandas.read_csv() 有一个 chunksize 可选参数。您可以使用它以较小的块处理文件。然后你可以使用df.to_sql() 将它们插入到表格中。
是的，我用 LOAD DATA INFILE 重写了我的代码。它运行良好，没有内存不足的问题现在我正在等待结果。谢谢你的时间@barmar

标签： mysql python-3.x pandas csv pyspark

【解决方案1】：

如果您确实想使用 Python 和 Pandas，而不是让 MySQL 直接读取 csv 文件的更快方法，请使用 Pandas 语法写入 sql。如果您可以将 csv 文件加载到内存中，那么它也应该可以加载到 MySQL 中。这是一个用于写入 MySQL 的代码的玩具示例，您可能需要许多选项，请参阅https://pandas.pydata.org/docs/reference/api/pandas.DataFrame.to_sql.html 了解详细信息。

df.to_sql(name=table, con=engine, if_exists='append', index=False, chunksize=10000)

【讨论】：

【解决方案2】：

你所做的可能有一些问题，无论如何我提供的都是一样的

首先，如果您尝试处理单行插入，可能 SQL 服务器无法选择请求的速度并崩溃，那么可能设置一个计时器设置敌人延迟会有所帮助

第二个，单个 INSERT 命令可以处理多行，因此您无需一次将一行推送到表中，而是可以以数千步的速度执行它们

我建议如下：

import time

rows = pd.DataFrame() #your df to be inserted

insert_header = "INSERT INTO table_name "
insert_cols = "(" + ",".join(rows.columns) + ")"

to_sql = rows[rows.columns[0:]].apply(
    lambda x: "('" + "','".join(x.fillna(0).astype(str)) + "'),",
    axis=1)

counter = to_sql.size
jump = 1000

for i in range(0, counter,jump): #jump
    print(jump)
    to = min(i + jump, counter)
    sql_values = " VALUES " + ''.join(to_sql.iloc[i:to])
    exec_sql = insert_header + insert_cols + sql_values
    exec_sql = exec_sql[:-1]
    cursor.execute(exec_sql)
    connection.commit()
    time.sleep(2)

【讨论】：