十大免费跨境网站,网络营销方案规范,WordPress网易云悬浮插件,做壁纸网站大家好#xff0c;使用Pandas和SQL高效地从数据库中读取、处理和写入大型数据集#xff0c;以实现最佳性能和内存管理#xff0c;这是十分重要的。 处理大型数据集往往是一项挑战#xff0c;特别是在涉及到从数据库读取和写入数据时。将整个数据集加载到内存中的传统方法可…大家好使用Pandas和SQL高效地从数据库中读取、处理和写入大型数据集以实现最佳性能和内存管理这是十分重要的。 处理大型数据集往往是一项挑战特别是在涉及到从数据库读取和写入数据时。将整个数据集加载到内存中的传统方法可能会导致系统崩溃和处理时间缓慢。
本文将探讨一种更好的解决方案简化分块读写数据的过程。这种技术能够高效地处理大量数据对于任何与数据库和数据帧一起工作的人来说都是一种宝贵的工具。我们将重点使用流行的数据分析库Pandas来演示如何从数据库表中读取大量数据并将其分块写入Pandas数据帧以及如何将大型数据从数据帧写回数据库。
一. 简化从数据库表中分块读取大型数据集的过程
在处理存储在数据库中的大量数据时以高效和可管理的方式处理数据非常重要。Pandas中的pd.read_sql()函数提供了一种方便的解决方案可以将数据从数据库表中读取到Pandas DataFrame中。通过添加chunksize参数可以控制每次加载到内存中的行数从而使我们能够以可管理的块处理数据并根据需要对其进行操作。本文将重点介绍如何使用Pandas从Postgres数据库中读取大型数据集。
engine create_engine(postgresqlpsycopg2://db_username:db_passworddb_host:db_port/db_name)conn engine.connect().execution_options(stream_resultsTrue)for chunk_dataframe in pd.read_sql(SELECT * FROM schema.table_name, conn, chunksize50000):print(fDataframe with {len(chunk_dataframe)} rows)# ...对数据帧做一些事情计算/操作...在上面的代码中 使用SQLAlchemy库中的create_engine()方法创建了一个SQLAlchemy引擎。 使用stream_resultsTrue创建了一个到PostgreSQL数据库的连接。稍后详细介绍。 然后将此连接与从表中选择所有行的SQL查询一起传递给pd.read_sql()函数。 还指定了chunksize为50000行这意味着pd.read_sql()函数每次返回一个包含50000行的新DataFrame。 然后可以使用for循环迭代pd.read_sql()函数返回的数据块。 在此示例中只是打印每个数据块中的行数但在真实场景中可能会在处理下一个数据块之前对每个数据块进行一些额外的处理。
stream_results在SQLAlchemy中当执行查询时通常会将结果一次性加载到内存中。当处理大型结果集时这可能会导致效率低下因为它需要大量的内存。当启用stream_results设置为True时查询会返回一个游标并在需要时获取结果集的每一行从而减少内存使用量。这在处理大型结果集时特别有用否则会占用大量内存。
二. 将大型数据集写入数据库表
在处理数据后可能需要将其写回数据库表。虽然Pandas提供的to_sql()方法是一种方便的方法但对于写入大量数据来说可能不是最高效的方法。我们将使用to_sql()的method参数。这时就要用到COPY方法。
COPY方法被广泛认为是将数据插入SQL数据库的最快方法之一。SQL中的COPY语句用于将大量数据快速加载到表中或将数据从文件导出到表中。COPY语句的基本语法简单明了可以轻松地将大量数据快速插入到数据库表中。
COPY [table_name] ([column1, column2, ...]) FROM [file_path] [WITH (options)]本文将探讨COPY方法以及它如何能够高效地将大量数据写入数据库表。无论处理的是少量数据还是大量数据COPY方法都是一个可以快速、高效地将数据写入数据库的有用工具。
在Python中一种方法是将数据帧存储在文件中然后使用上述查询快速批量插入数据。但是大多数情况下并不希望创建文件因此我们将使用缓冲对象。 注意此方法仅适用于支持COPY FROM方法的数据库。 import csv
from io import StringIOdef copy_insert(table, conn, keys, data_iter):# 获取提供游标的DBAPI连接dbapi_conn conn.connectionwith dbapi_conn.cursor() as cur:string_buffer StringIO()writer csv.writer(string_buffer)writer.writerows(data_iter)string_buffer.seek(0)columns , .join([{}.format(k) for k in keys])if table.schema:table_name {}.{}.format(table.schema, table.name)else:table_name table.namesql COPY {} ({}) FROM STDIN WITH CSV.format(table_name, columns)cur.copy_expert(sqlsql, filestring_buffer)现在来解读一下上面的代码
1. copy_insert函数是一个实用函数使用COPY FROM方法将数据插入数据库表中这是一种比标准INSERT语句更快的插入数据方法。
2. 该函数需要四个参数 table是代表数据库中表的pandas.io.sql.SQLTable对象。 conn是连接到数据库的SQLAlchemy连接对象。 keys是列名列表。 data_iter是提供要插入的值的可迭代对象。
3. 该函数首先从SQLAlchemy连接对象获取一个DBAPI连接并创建一个游标。
4. 然后将要插入的值以CSV文件的形式写入到StringIO缓冲区中并将其传递给游标的copy_expert方法。 copy_expert方法用于执行COPY语句将CSV文件中的数据插入数据库表中。table_name变量可以通过使用模式名称和表名称或仅使用表名称来构造这取决于表是否定义了模式例如MySQL没有模式而PostgreSQL有模式。
5. 使用SQL参数执行COPY语句并将文件缓冲区作为文件参数插入数据到数据库中。
为了插入数据将使用SQLAlchemy的基本方法
df.to_sql(nametable_name, schemaschema_name, conengine, if_existsappend, indexFalse, methodcopy_insert)name数据库中表格的名称。 schema表所属数据库模式的名称。 conSQLAlchemy引擎对象表示与数据库的连接。 if_exists一个字符串用于指定如果表已经存在时的行为在本例中为append。使用append时新行将被添加到现有表中。 index一个布尔值指定是否将DataFrame索引作为表中的单独列写入本例中为False。 method一个字符串用于指定向表中写入数据的方法。我们将使用前面定义的copy_insert。
接下来数据将快速、高效地插入数据库表中。
