pandas.DataFrame.to_sql #
- 数据框。to_sql ( name , con , * , schema = None , if_exists = 'fail' , index = True , index_label = None , chunksize = None , dtype = None , method = None ) [来源] #
将存储在 DataFrame 中的记录写入 SQL 数据库。
支持 SQLAlchemy [1]支持的数据库。可以新建、追加或覆盖表。
- 参数:
- 名称字符串
SQL 表的名称。
- con sqlalchemy.engine.(引擎或连接)或 sqlite3.Connection
使用 SQLAlchemy 可以使用该库支持的任何数据库。为 sqlite3.Connection 对象提供旧版支持。用户负责 SQLAlchemy 可连接的引擎处理和连接关闭。看这里。如果传递已经在事务中的 sqlalchemy.engine.Connection,则该事务将不会被提交。如果传递 sqlite3.Connection,则无法回滚记录插入。
- schema str,可选
指定模式(如果数据库风格支持此模式)。如果没有,则使用默认架构。
- if_exists {'失败', '替换', '追加'}, 默认'失败'
如果表已经存在,该如何处理。
失败:引发 ValueError。
替换:在插入新值之前删除表。
追加:将新值插入到现有表中。
- 索引bool,默认 True
将 DataFrame 索引写为列。使用index_label作为表中的列名。为此列创建表索引。
- index_label str 或序列,默认 None
索引列的列标签。如果没有给出(默认)并且 索引为True,则使用索引名称。如果 DataFrame 使用 MultiIndex,则应给出序列。
- chunksize int,可选
指定每个批次中一次写入的行数。默认情况下,将立即写入所有行。
- dtype字典或标量,可选
指定列的数据类型。如果使用字典,键应该是列名,值应该是 SQLAlchemy 类型或 sqlite3 传统模式的字符串。如果提供标量,它将应用于所有列。
- 方法{无,'多',可调用},可选
控制使用的 SQL 插入子句:
None :使用标准 SQL
INSERT
子句(每行一个)。'multi':在单个
INSERT
子句中传递多个值。可通过签名调用。
(pd_table, conn, keys, data_iter)
详细信息和示例可调用实现可以在插入方法部分中找到。
- 返回:
- 无或整数
受 to_sql 影响的行数。如果传入的可调用函数
method
不返回整数行,则不返回任何内容。受影响的返回行数是 SQLAlchemy或 SQLAlchemy 可连接
rowcount
属性的总和,这可能无法反映sqlite3或 SQLAlchemysqlite3.Cursor
中规定的确切写入行数 。1.4.0 版本中的新增功能。
- 加薪:
- 值错误
当表已存在且if_exists为“失败”(默认值)时。
也可以看看
read_sql
从表中读取 DataFrame。
笔记
如果数据库支持,时区感知日期时间列将被写入 SQLAlchemy 的类型。否则,日期时间将存储为原始时区本地的时区未知时间戳。
Timestamp with timezone
并非所有数据存储都支持
method="multi"
.例如,Oracle 不支持多值插入。参考
例子
创建内存 SQLite 数据库。
>>> from sqlalchemy import create_engine >>> engine = create_engine('sqlite://', echo=False)
从头开始创建一个包含 3 行的表。
>>> df = pd.DataFrame({'name' : ['User 1', 'User 2', 'User 3']}) >>> df name 0 User 1 1 User 2 2 User 3
>>> df.to_sql(name='users', con=engine) 3 >>> from sqlalchemy import text >>> with engine.connect() as conn: ... conn.execute(text("SELECT * FROM users")).fetchall() [(0, 'User 1'), (1, 'User 2'), (2, 'User 3')]
sqlalchemy.engine.Connection也可以传递给con :
>>> with engine.begin() as connection: ... df1 = pd.DataFrame({'name' : ['User 4', 'User 5']}) ... df1.to_sql(name='users', con=connection, if_exists='append') 2
这可以支持需要在整个操作中使用相同 DBAPI 连接的操作。
>>> df2 = pd.DataFrame({'name' : ['User 6', 'User 7']}) >>> df2.to_sql(name='users', con=engine, if_exists='append') 2 >>> with engine.connect() as conn: ... conn.execute(text("SELECT * FROM users")).fetchall() [(0, 'User 1'), (1, 'User 2'), (2, 'User 3'), (0, 'User 4'), (1, 'User 5'), (0, 'User 6'), (1, 'User 7')]
仅用 覆盖表
df2
。>>> df2.to_sql(name='users', con=engine, if_exists='replace', ... index_label='id') 2 >>> with engine.connect() as conn: ... conn.execute(text("SELECT * FROM users")).fetchall() [(0, 'User 6'), (1, 'User 7')]
用于
method
定义可调用插入方法,以便在 PostgreSQL 数据库中的表上存在主键冲突时不执行任何操作。>>> from sqlalchemy.dialects.postgresql import insert >>> def insert_on_conflict_nothing(table, conn, keys, data_iter): ... # "a" is the primary key in "conflict_table" ... data = [dict(zip(keys, row)) for row in data_iter] ... stmt = insert(table.table).values(data).on_conflict_do_nothing(index_elements=["a"]) ... result = conn.execute(stmt) ... return result.rowcount >>> df_conflict.to_sql(name="conflict_table", con=conn, if_exists="append", method=insert_on_conflict_nothing) 0
对于 MySQL,可调用以更新列
b
以及c
主键上是否存在冲突。>>> from sqlalchemy.dialects.mysql import insert >>> def insert_on_conflict_update(table, conn, keys, data_iter): ... # update columns "b" and "c" on primary key conflict ... data = [dict(zip(keys, row)) for row in data_iter] ... stmt = ( ... insert(table.table) ... .values(data) ... ) ... stmt = stmt.on_duplicate_key_update(b=stmt.inserted.b, c=stmt.inserted.c) ... result = conn.execute(stmt) ... return result.rowcount >>> df_conflict.to_sql(name="conflict_table", con=conn, if_exists="append", method=insert_on_conflict_update) 2
指定数据类型(对于缺少值的整数特别有用)。请注意,虽然 pandas 被迫将数据存储为浮点型,但数据库支持可为空的整数。当使用 Python 获取数据时,我们返回整数标量。
>>> df = pd.DataFrame({"A": [1, None, 2]}) >>> df A 0 1.0 1 NaN 2 2.0
>>> from sqlalchemy.types import Integer >>> df.to_sql(name='integers', con=engine, index=False, ... dtype={"A": Integer()}) 3
>>> with engine.connect() as conn: ... conn.execute(text("SELECT * FROM integers")).fetchall() [(1,), (None,), (2,)]