如何使用索引加速 SQL 查询 [Python 版]

推荐：使用 NSDT场景编辑器助你快速搭建可二次编辑器的3D应用场景

假设您正在筛选一本书的页面。而且您希望更快地找到所需的信息。你是怎么做到的？好吧，您可能会查找术语索引，然后跳转到引用特定术语的页面。SQL 中的索引的工作方式与书籍中的索引类似。

在大多数实际系统中，您将针对具有大量行（例如数百万行）的数据库表运行查询。需要全表扫描所有行以检索结果的查询将非常慢。如果您知道必须经常基于某些列查询信息，则可以在这些列上创建数据库索引。这将大大加快查询速度。

那么我们今天会学到什么呢？我们将学习使用 sqlite3 模块在 Python 中连接和查询 SQLite 数据库。我们还将学习如何添加索引，并了解它如何提高性能。

要按照本教程编写代码，您应该在工作环境中安装 Python 3.7+ 和 SQLite。

注意：本教程中的示例和示例输出适用于 Ubuntu LTS 3.10 上的 Python 3.3 和 SQLite37（版本 2.22.04）。

在 Python 中连接到数据库

我们将使用内置的 sqlite3 模块。在开始运行查询之前，我们需要：

• 连接到数据库
• 创建数据库游标以运行查询

若要连接到数据库，我们将使用

来自 sqlite3 模块的 connect（） 函数。建立连接后，我们可以调用连接对象来创建数据库游标，如下所示：cursor()

import sqlite3# connect to the db
db_conn = sqlite3.connect('people_db.db')
db_cursor = db_conn.cursor()

在这里，我们尝试连接到数据库

people_db.如果数据库不存在，运行上面的代码片段将为我们创建 sqlite 数据库。

创建表并插入记录

现在，我们将在数据库中创建一个表，并用记录填充它。

让我们在数据库中创建一个名为 people 的表，其中包含以下字段：people_db

• 名字
• 电子邮件
• 工作

# main.py
...
# create table
db_cursor.execute('''CREATE TABLE people (id INTEGER PRIMARY KEY,name TEXT,email TEXT,job TEXT)''')...# commit the transaction and close the cursor and db connection
db_conn.commit()
db_cursor.close()
db_conn.close()

使用伪造器生成合成数据

我们现在必须在表中插入记录。为此，我们将使用 Faker——一个用于合成数据生成的 Python 包——可通过 pip 安装：

$ pip install faker

安装faker后，可以将类导入到Python脚本中：Faker

# main.py
...
from faker import Faker
...

下一步是生成记录并将其插入人员表。为了让我们知道索引如何加快查询速度，让我们插入大量记录。在这里，我们将插入 100K 条记录;将变量设置为 100000。num_records

然后，我们执行以下操作：

• 实例化一个对象并设置种子，以便我们获得可重现性。Fakerfake
• 使用名字和姓氏获取名称字符串 - 通过调用对象和对象。first_name()last_name()fake
• 通过调用生成假域。domain_name()
• 使用名字和姓氏以及域生成电子邮件字段。
• 使用 获取每个单独记录的作业。job()

我们生成记录并将其插入到表中：people

# create and insert records
fake = Faker() # be sure to import: from faker import Faker 
Faker.seed(42)num_records = 100000for _ in range(num_records):first = fake.first_name()last = fake.last_name()name = f"{first} {last}"domain = fake.domain_name()email = f"{first}.{last}@{domain}"job = fake.job()db_cursor.execute('INSERT INTO people (name, email, job) VALUES (?,?,?)', (name,email,job))# commit the transaction and close the cursor and db connection
db_conn.commit()
db_cursor.close()
db_conn.close()

现在，main.py 文件具有以下代码：

# main.py
# imports
import sqlite3
from faker import Faker# connect to the db
db_conn = sqlite3.connect('people_db.db')
db_cursor = db_conn.cursor()# create table
db_cursor.execute('''CREATE TABLE people (id INTEGER PRIMARY KEY,name TEXT,email TEXT,job TEXT)''')# create and insert records
fake = Faker()
Faker.seed(42)num_records = 100000for _ in range(num_records):first = fake.first_name()last = fake.last_name()name = f"{first} {last}"domain = fake.domain_name()email = f"{first}.{last}@{domain}"job = fake.job()db_cursor.execute('INSERT INTO people (name, email, job) VALUES (?,?,?)', (name,email,job))# commit the transaction and close the cursor and db connection
db_conn.commit()
db_cursor.close()
db_conn.close()

运行此脚本一次，以使用记录数填充表。num_records

查询数据库

现在我们有了包含 100K 条记录的表，让我们对表运行一个示例查询。people

让我们运行一个查询来：

• 获取职位名称为“产品经理”的记录的名称和电子邮件，以及
• 将查询结果限制为 10 条记录。

我们将使用 time 模块中的默认计时器来获取查询的大致执行时间。

# sample_query.pyimport sqlite3
import timedb_conn = sqlite3.connect("people_db.db")
db_cursor = db_conn.cursor()t1 = time.perf_counter_ns()db_cursor.execute("SELECT name, email FROM people WHERE job='Product manager' LIMIT 10;")res = db_cursor.fetchall()
t2 = time.perf_counter_ns()print(res)
print(f"Query time without index: {(t2-t1)/1000} us")

下面是输出：

Output >>
[("Tina Woods", "[e-mail]"),("Toni Jackson", "[e-mail]"),("Lisa Miller", "[e-mail]"),("Katherine Guerrero", "[e-mail]"),("Michelle Lane", "[e-mail]"),("Jane Johnson", "[e-mail]"),("Matthew Odom", "[e-mail]"),("Isaac Daniel", "[e-mail]"),("Jay Byrd", "[e-mail]"),("Thomas Kirby", "[e-mail]"),
]Query time without index: 448.275 us

您还可以通过在命令行运行来调用 SQLite 命令行客户端：sqlite3 db_name

$ sqlite3 people_db.db
SQLite version 3.37.2 2022-01-06 13:25:41
Enter ".help" for usage hints.

要获取索引列表，您可以运行：.index

sqlite> .index

由于当前没有索引，因此不会列出任何索引。

您还可以像这样检查查询计划：

sqlite> EXPLAIN QUERY PLAN SELECT name, email FROM people WHERE job='Product Manager' LIMIT 10;
QUERY PLAN
`--SCAN people

这里的查询计划是扫描所有效率低下的行。

在特定列上创建索引

若要在特定列上创建数据库索引，可以使用以下语法：

CREATE INDEX index-name on table (column(s))

假设我们需要经常查找具有特定职位的个人的记录。在作业列上创建索引会有所帮助：people_job_index

# create_index.pyimport time
import sqlite3db_conn = sqlite3.connect('people_db.db')db_cursor =db_conn.cursor()t1 = time.perf_counter_ns()db_cursor.execute("CREATE INDEX people_job_index ON people (job)")t2 = time.perf_counter_ns()db_conn.commit()print(f"Time to create index: {(t2 - t1)/1000} us")Output >>
Time to create index: 338298.6 us

尽管创建索引需要这么长时间，但这是一次性操作。运行多个查询时，您仍将获得显著的加速。

现在，如果您在 SQLite 命令行客户端上运行，您将获得：.index

sqlite> .index
people_job_index

使用索引查询数据库

如果您现在查看查询计划，您应该能够看到我们现在使用作业列上的索引搜索表：peoplepeople_job_index

sqlite> EXPLAIN QUERY PLAN SELECT name, email FROM people WHERE job='Product manager' LIMIT 10;
QUERY PLAN
`--SEARCH people USING INDEX people_job_index (job=?)

您可以重新运行sample_query.py。仅修改语句并查看查询现在运行需要多长时间：print()

# sample_query.pyimport sqlite3
import timedb_conn = sqlite3.connect("people_db.db")
db_cursor = db_conn.cursor()t1 = time.perf_counter_ns()db_cursor.execute("SELECT name, email FROM people WHERE job='Product manager' LIMIT 10;")res = db_cursor.fetchall()
t2 = time.perf_counter_ns()print(res)
print(f"Query time with index: {(t2-t1)/1000} us")

下面是输出：

Output >>
[("Tina Woods", "[e-mail]"),("Toni Jackson", "[e-mail]"),("Lisa Miller", "[e-mail]"),("Katherine Guerrero", "[e-mail]"),("Michelle Lane", "[e-mail]"),("Jane Johnson", "[e-mail]"),("Matthew Odom", "[e-mail]"),("Isaac Daniel", "[e-mail]"),("Jay Byrd", "[e-mail]"),("Thomas Kirby", "[e-mail]"),
]Query time with index: 167.179 us

我们看到查询现在大约需要 167.179 微秒来执行。

性能改进

对于我们的示例查询，使用 index 进行查询的速度大约快 2.68 倍。我们在执行时间中获得了 62.71% 的百分比加速。

您还可以尝试运行更多查询：涉及对作业列进行筛选并查看性能改进的查询。

另请注意：由于我们仅在作业列上创建了索引，因此，如果您运行的查询涉及其他列，则查询的运行速度不会比没有索引时快。

总结和后续步骤

我希望本指南能帮助您了解在频繁查询的列上创建数据库索引如何显著加快查询速度。这是对数据库索引的介绍。您还可以创建多列索引、同一列的多个索引等等。

原文链接：如何使用索引加速 SQL 查询 [Python 版] (mvrlink.com)