Python库pandas之六

输入/输出

read_sql

函数

词法：pandas.read_sql(sql, con, index_col=None, coerce_float=True, params=None, parse_dates=None, columns=None, chunksize=None, dtype_backend=<no_default>, dtype=None)

read_sql将SQL查询，或数据库表读入DataFrame。

该函数是 read_sql_table 和 read_sql_query 的便捷包装器（为了向后兼容）。它将根据提供的输入委托给特定的函数。 SQL 查询将被路由到 read_sql_query，而数据库表名称将被路由到 read_sql_table。

参数说明

• sql，该参数类型是字符串，或SQLAlchemy Selectable
  要执行的 SQL 查询，或table名。

• con，该参数类型是ADBC连接, SQLAlchemy连接, 字符串, 或sqlite3连接
  ADBC 提供具有主机支持的高性能 I/O。
  SQLAlchemy 可以使用该库支持的任何数据库。
  如果是 DBAPI2 对象，则仅支持 sqlite3。
  用户负责 ADBC 连接和 SQLAlchemy连接的引擎处置和连接关闭； 字符串连接会自动关闭。

• index_col，该参数类型是字符串，或字符串list, 是可选的, 默认值为None
  index_col参数是设置为索引的列（MultiIndex）。

• coerce_float，该参数类型是bool, 默认值为True
  尝试将非字符串、非数字对象（如decimal.Decimal）的值转换为浮点型，这对于SQL结果集很有用。

• params，该参数类型是list, 元组tuple，或字典dict, 是可选的, 默认值为None
  要传递给数据库的参数列表。用于传递参数的语法取决于数据库驱动程序。检查数据库驱动程序文档，了解支持的语法样式中。

• parse_dates，该参数类型是list，或字典dict, 默认值为None
  parse_dates指定要作为日期分析的列名称表。
  {column_name: format} 的字典，在解析字符串时间时，其中format与strftime 兼容，或者，在解析整数时间戳的情况下，是 (D, s, ns, ms, us) 之一。
  {column_name: arg} 的字典，其中arg是一个字典，对应于pandas.to_datetime() 的关键字参数，对于没有本机日期时间支持的数据库（例如 SQLite）特别有用。

• columns，该参数类型是list, 默认值为None
  从SQL表中选择的列名表。

• chunksize，该参数类型是int, 默认值为None
  如果在read_sql调用中指定该参数，则read_sql返回一个迭代器，其中chunksize是每个块中包含的行数。

• dtype_backend，该参数类型是{‘numpy_nullable’, ‘pyarrow’}, 默认值为‘numpy_nullable’
  应用于生成的DataFrame的后端数据类型。行为如下：
  “numpy_nullable”：返回支持 nullable-dtype 的 DataFrame。
  “pyarrow”：返回 pyarrow 支持的可为空的 ArrowDtype的DataFrame。

• dtype，该参数类型是Type名称，或列字典
  数据或列的数据类型。例如。 np.float64 或 {‘a’: np.float64, ‘b’: np.int32, ‘c’: ‘Int64’}。如果传递的是表而不是查询，则忽略该参数。

应用实列

下列产生sqlite3数据库的table, 读csv数据加入数据库，读取数据库table

import sys
import sqlite3 as sqlite
import pandas as pd
from sqlalchemy import create_engineDB_NAME = "test_db.db"class sqlite3_db():def __init__(self, dbn):self.dbn = dbndef db_connect(self):self.con = sqlite.connect(self.dbn)def db_query(self):self.con = sqlite.connect(self.dbn)cur = self.con.cursor()rows = cur.execute("SELECT * FROM peoples")print(rows.fetchall())def db_add_records(self, df):self.db_connect()data = []for rec in df.values:col_vals = list(rec.flatten())col_names = ("name", "age", "salary")d1 = dict(zip(col_names, col_vals))data.append(d1)self.con.executemany("INSERT INTO peoples VALUES(:name, :age, :salary)", data)self.con.commit()self.con.close()def db_create_table(self):self.db_connect()cur = self.con.execute("CREATE TABLE peoples(name, age, salary)")data = ({"name":"john", "age":35, "salary":5000.90}, {"name":"Tom", "age":36, "salary":4984},)self.con.executemany("INSERT INTO peoples VALUES(:name, :age, :salary)", data)self.con.commit()self.con.close()if __name__ == "__main__":if len(sys.argv) == 1:sql = sqlite3_db(DB_NAME)sql.db_query()     elif  sys.argv[1] == '0':sql = sqlite3_db(DB_NAME)sql.db_create_table()elif sys.argv[1] == '1':cnx = create_engine('sqlite:///test_db.db').connect()df = pd.read_sql_table('peoples', cnx)print(df)elif sys.argv[1] == '2':con = sqlite.connect('test_db.db')sql = "SELECT * FROM peoples"df = pd.read_sql(sql, con)print(df)l = list(df['age'])print("type: {0}".format(type(l)))elif sys.argv[1] == '3':if len(sys.argv) == 3:df = pd.read_csv(sys.argv[2])sql = sqlite3_db(DB_NAME)sql.db_add_records(df)print(df)

文件data_csv_1.txt的内容

name, age, salary
Helen, 25, 40000
Michael, 20, 20000

程序运行的屏幕输出

C:\>python sqlite3_1.py 0
C:\>python sqlite3_1.py 1name  age  salary
0  john   35  5000.9
1   Tom   36  4984.0C:\>python sqlite3_1.py 2name  age  salary
0  john   35  5000.9
1   Tom   36  4984.0
type: <class 'list'>C:\>python sqlite3_1.py 3 data_csv_1.txtname   age   salary
0    Helen    25    40000
1  Michael    20    20000C:\>python sqlite3_1.py 2name  age   salary
0     john   35   5000.9
1      Tom   36   4984.0
2    Helen   25  40000.0
3  Michael   20  20000.0
type: <class 'list'>

下列例子说明如何使用chunksize

import sys
import sqlite3 as sqlite
import pandas as pd
from sqlalchemy import create_engineDB_NAME = "test_db.db"class sqlite3_db():def __init__(self, dbn):self.dbn = dbndef db_sql_records(self, csv_fn):df = pd.read_csv(csv_fn)self.con = sqlite.connect(self.dbn)data = []for rec in df.values:col_vals = list(rec.flatten())col_names = ("name", "age", "salary")d1 = dict(zip(col_names, col_vals))data.append(d1)cur = self.con.execute("CREATE TABLE peoples(name, age, salary)")self.con.executemany("INSERT INTO peoples VALUES(:name, :age, :salary)", data)self.con.commit()sql = "SELECT * FROM peoples"df = pd.read_sql(sql, self.con)print(df)it = pd.read_sql(sql, self.con, chunksize = 2)for x in it:print()print(x)self.con.close()if __name__ == "__main__":sql = sqlite3_db(DB_NAME)if len(sys.argv) > 1:sql.db_sql_records(sys.argv[1])

程序运行的屏幕输出

C:\>python sqlite3_2.py data_csv_1.txtname  age  salary
0    Helen   25   40000
1  Michael   20   20000
2      Tom   35   40000
3     John   30   20000
4     Jeff   25   50000
5  Johnson  350  120000name  age  salary
0    Helen   25   40000
1  Michael   20   20000name  age  salary
0   Tom   35   40000
1  John   30   20000name  age  salary
0     Jeff   25   50000
1  Johnson  350  120000

文件data_csv_1.txt的内容

name, age, salary
Helen, 25, 40000
Michael, 20, 20000
Tom, 35, 40000
John, 30, 20000
Jeff, 25, 50000
Johnson, 350, 120000