# 使用 mysql-connector-python 操作 TiDB

## 概述

TiDB 是一个分布式、云原生、水平可扩展的数据库，支持混合事务和分析处理（HTAP）。对于应用程序开发人员来说，TiDB 是一个始终可用、可靠且可无限扩展的数据库。

+ 在本实验中，您将学习如何：
  + 使用 Python 代码连接、访问和修改 TiDB Cloud Serverless 集群中的数据。
  + 使用预处理语句和批量插入来提高应用程序的性能。
  + 使用 `AUTO_RANDOM` 替代 `AUTO_INCREMENT` 以避免热点问题。
  + 使用在线 DDL 向表中添加列，而不会阻塞或中断正在进行的事务。
  + 使用和比较悲观和乐观模式进行并发控制。

### 时长

+ 完成本实验大约需要 90 分钟。

> [!NOTE]
> 在本实验中，您将使用基于 Web 的 VS Code。使用基于 Web 的 IDE 在 TiDB 中并非强制性要求。但是，这在云数据库协作中是一种常见的方法。通过将开发者客户端部署在靠近云数据库的位置，我们可以更好地模拟应用服务器的真实行为和网络条件。

> [!TIP]
> 实验环境部署完成后，点击底部栏的 **前进** 开始 **模块 1**。

## 模块 1：准备运行 Python 代码

### 步骤 1. 在 IDE 中打开终端

+ 在浏览器中打开您的 [VS Code](${LAB:SECURITY_ACCESS_PROXY_URL}?next=${LAB:"/vscode/?folder=/root/tidb-lab-python"|urlencode})。
+ 在弹出窗口中点击 `Yes, I trust the authors`。

  ![90%](https://lab-static.pingcap.com/quick-demo/vscode-trust.jpg)

+ 点击 `≡`，`Terminal`，然后点击 `New Terminal` 在 VS Code 中打开一个新终端。

  ![90%](https://public-pingcap-innovation.s3.us-west-2.amazonaws.com/demo-405/405-open-terminal-0307.jpg)

### 步骤 2. 安装 Python 依赖

+ 在 VS Code 终端中运行以下命令来安装所需的依赖。

  ```bash
  poetry install
  ```

  ```output
  Creating virtualenv tidb-lab-python-CQ_Jo6Zv-py3.11 in /root/.cache/pypoetry/virtualenvs
  Installing dependencies from lock file

  Package operations: 4 installs, 0 updates, 0 removals

    - Installing python-dotenv (1.0.1)
    - Installing dotenv (0.9.9)
    - Installing mysql-connector-python (9.2.0)
    - Installing pymysql (1.1.1)
  ```

### 步骤 3. 创建 .env 文件

+ 在 `tidb-lab-python/` 文件夹下创建一个 `.env` 文件。

  ![90%](https://public-pingcap-innovation.s3.us-west-2.amazonaws.com/demo-405/405-new-env-0307.jpg)

+ 将以下内容添加到 `.env` 文件中并保存。

  ```properties
  TIDB_HOST=${LAB:SERVERLESS_CLUSTER_HOST}
  TIDB_PORT=${LAB:SERVERLESS_CLUSTER_PORT}
  TIDB_USER='${LAB:SERVERLESS_CLUSTER_USERNAME}'
  TIDB_PASSWORD='${LAB:SERVERLESS_CLUSTER_PASSWORD}'
  TIDB_DB_NAME=${LAB:SERVERLESS_CLUSTER_DATABASE_NAME}
  ```

## 模块 2：运行 Python 代码

运行 Python 代码以连接到 TiDB Cloud Serverless 集群并执行一些基本操作。

### 步骤 1. 连接到 TiDB Cloud Serverless 集群

+ 在 VS Code 左侧的 **EXPLORER** 中，在 `mysql_connector_python_demo` 文件夹下，点击 `demo_connection.py` 文件，您可以简单浏览源代码。这个 Python 脚本将连接到实验用的 TiDB Cloud Serverless 集群并创建一个名为 `${LAB:SERVERLESS_CLUSTER_DATABASE_NAME}` 的新数据库。

+ 在 VS Code 终端中运行 `demo_connection.py`。

  ```bash
  poetry run python mysql_connector_python_demo/demo_connection.py
  ```

  ```output
  Connected to gateway01.us-west-2.prod.aws.tidbcloud.com:4000 as 42yK4xxxxxxxxxxxxsP (Connection ID: 1233194)
  The server is running 5.7.28-TiDB-Serverless
  ```

  ![90%](https://public-pingcap-innovation.s3.us-west-2.amazonaws.com/demo-405/405-demo-connection-0307.jpg)

> [!IMPORTANT]
> 连接字符串存储在 `.env` 文件中。
> 我们将在后续步骤中使用 `demo_connection.py` 中的 `get_connection` 函数来获取到 TiDB Cloud Serverless 集群的连接。

### 步骤 2. 插入数据

+ 在 VS Code 左侧的 **EXPLORER** 中，在 `mysql_connector_python_demo` 文件夹下，点击 `positional_placeholders.py` 文件，您可以简单浏览源代码。

+ 通过位置占位符插入数据。

  ```bash
  poetry run python mysql_connector_python_demo/positional_placeholders.py
  ```

  ```output
  1 row inserted into table t1 by positional placeholders
  ```

  ![90%](https://public-pingcap-innovation.s3.us-west-2.amazonaws.com/demo-405/405-positional-placeholders-0307.jpg)

+ 在 VS Code 左侧的 **EXPLORER** 中，在 `mysql_connector_python_demo` 文件夹下，点击 `named_placeholders.py` 文件，您可以简单浏览源代码。

+ 通过命名占位符插入数据。

  ```bash
  poetry run python mysql_connector_python_demo/named_placeholders.py
  ```

  ```output
  1 row inserted into table t1 by named placeholders
  ```

  ![90%](https://public-pingcap-innovation.s3.us-west-2.amazonaws.com/demo-405/405-named-placeholders-0307.jpg)

### 步骤 3. 查询数据

+ 在 VS Code 左侧的 **EXPLORER** 中，在 `mysql_connector_python_demo` 文件夹下，点击 `query_fetch_all.py` 文件，您可以简单浏览源代码。

+ 使用 `fetchall()` 函数查询数据。

  ```bash
  poetry run python mysql_connector_python_demo/query_fetch_all.py
  ```

  ```output
  PK of ABC is 1
  PK of DEF is 2
  ```

  ![90%](https://public-pingcap-innovation.s3.us-west-2.amazonaws.com/demo-405/405-fetchall-0307.jpg)

+ 在 VS Code 左侧的 **EXPLORER** 中，在 `mysql_connector_python_demo` 文件夹下，点击 `query_fetch_one.py` 文件，您可以简单浏览源代码。

+ 使用 `fetchone()` 函数查询数据。

  ```bash
  poetry run python mysql_connector_python_demo/query_fetch_one.py
  ```

  ```output
  PK of ABC is 1
  PK of DEF is 2
  ```

  ![90%](https://public-pingcap-innovation.s3.us-west-2.amazonaws.com/demo-405/405-fetchone-0307.jpg)

## 模块 3：使用预处理语句和批量插入

现在，您可以使用 Python 代码读写 TiDB Cloud Serverless 集群中的数据。但如果您想提高应用程序的性能，可以使用预处理语句（用于预编译和重复执行）和高效的批量插入（用于在单个 SQL 语句中向表中插入多行数据）。

### 步骤 1. 使用预处理语句

+ 在 VS Code 左侧的 **EXPLORER** 中，在 `mysql_connector_python_demo` 文件夹下，点击 `prepared_statement.py` 文件，查看示例代码中的 `connection.cursor(prepared=True)` 和 `print("Generated PK is", cursor.lastrowid)` 行。如果您使用参数 `prepared=False` 创建游标，则只使用客户端预处理语句。`cursor.lastrowid` 用于获取最近插入行的自动随机 ID。

+ 运行 `prepared_statement.py`。

  ```bash
  poetry run python mysql_connector_python_demo/prepared_statement.py
  ```
  ```output
  The generated PK is 4035225266123964417
  ```

![90%](https://public-pingcap-innovation.s3.us-west-2.amazonaws.com/demo-405/405-prepared-statement-0307.jpg)

### 步骤 2. 使用批量插入

+ 然后，我们将比较批量插入和单行插入的性能。在 VS Code 左侧的 **EXPLORER** 中，在 `mysql_connector_python_demo` 文件夹下，点击 `batch_insert.py` 文件。

+ 在这一步中，我们将向表 `t1_batchtest` 插入 1000 行数据。在批量插入函数 `_batch_style_insert` 中，我们将使用 `executemany` 函数插入数据。另一方面，我们将在单行插入函数 `_single_style_insert` 中使用 `execute` 函数逐行插入数据。

  ```bash
  poetry run python mysql_connector_python_demo/batch_insert.py
  ```
  ```output
  Batch Inserting 1000 rows in 84.9921875 (ms).
  Total rows in t1_batchtest table: 1000.
  Non-Batch Inserting 1000 rows in 19608.541259765625 (ms).
  Total rows in t1_batchtest table: 1000.
  ```

![90%](https://public-pingcap-innovation.s3.us-west-2.amazonaws.com/demo-405/405-batch-insert-0307.jpg)

+ 这里我们可以看到，批量插入比单行插入快得多。

## 模块 4：使用 AUTO_RANDOM 替代 AUTO_INCREMENT 以避免热点

+ 在像 TiDB 这样的分布式数据库中，数据被分成多个 Region 并分布在多个存储节点（TiKV 节点）上。

+ `AUTO_INCREMENT` 问题：
  + 使用 `AUTO_INCREMENT` 时，ID 是连续的（1, 2, 3, 4, ...）
  + 新插入的数据都会进入同一个 Region
  + 这会在一个存储节点处理大部分写入流量时，造成"热点"
  + 其他存储节点未被充分利用
  + 这种瓶颈会严重限制写入吞吐量

+ 解决方案 - `AUTO_RANDOM`：
  + `AUTO_RANDOM` 生成非连续的、分布式的 ID
  + 这些分布式 ID 会使得记录写入不同的 Region
  + INSERT 工作负载在所有存储节点之间均衡
  + 这提高了高并发场景下的写入吞吐量和可扩展性

  ```bash
  poetry run python mysql_connector_python_demo/auto_random.py
  ```
  ```output
  ...
  AUTO_RANDOM IDs (distributed to avoid hotspots):
  3
  288230376151711748
  1441151880758558729
  1729382256910270465
  3458764513820540934
  3458764513820540935
  4899916394579099650
  5764607523034234888
  6917529027641081866
  7782220156096217093

  AUTO_INCREMENT IDs (sequential, can cause hotspots):
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  ...
  ```

## 模块 5：使用在线 DDL 执行在线表结构变更

+ 本模块演示如何在 **TiDB** 中执行**在线表结构变更（DDL 操作）**，同时确保正在运行的**INSERT 任务**继续运行而不中断。这个过程中的一个关键挑战是处理**错误代码 8028**，这种错误发生在表结构变更暂时阻塞事务时。演示展示了一种处理这种情况的稳健方法，允许插入作业在表结构变更期间恢复并继续运行。

+ 本模块包含两个主要操作：

1. **插入作业（`--insert`）**
   + 持续向指定表中插入新记录。
   + 处理可能由于并发 DDL 操作而发生的表结构变更异常（错误代码 **8028**）。
   + 在失败时重试插入操作以维持服务可用性。

2. **表结构修改（`--alter`）**
   + 在插入进行时执行 `ALTER TABLE` 命令以修改表结构。
   + 演示 TiDB 允许**在线表结构变更**，但事务可能会因表结构变更而暂时失败。
   + 确保新表结构成功应用，同时进行中的事务能够相应调整。

> [!TIP]
> **关键实现细节：**
>
> + **处理 TiDB 元数据锁**：脚本**禁用元数据锁定**（`SET GLOBAL tidb_enable_metadata_lock = OFF`）以允许 DDL 变更而不阻塞查询。
> + **自动错误处理**：插入作业**检测错误代码 8028** 并**重试失败的事务**以防止崩溃。
> + **并发执行**：脚本允许在单独的终端中运行**插入作业**和**表结构修改**，模拟数据库更新与进行中事务并存的真实场景。

+ 首先运行插入作业。

  ```bash
  poetry run python mysql_connector_python_demo/online_ddl.py --insert
  ```
  ```output
  Starting insert job on table online_ddl_test
  You can now run ALTER TABLE in another terminal with:
  python online_ddl.py --alter online_ddl_test
  Connected to TiDB: 2nR9s2Y6oh4rkp7.6z3VQEZ4@gateway01.us-west-2.prod.aws.tidbcloud.com:4000 using database test
  Disabled TiDB metadata lock for online DDL operations
  Current setting: tidb_enable_metadata_lock = OFF
  Created table online_ddl_test
  Inserting record 0
  Inserting record 1
  Inserting record 2
  Inserting record 3
  Inserting record 4
  Inserting record 5
  ...
  ```

  ![90%](https://public-pingcap-innovation.s3.us-west-2.amazonaws.com/demo-405/405-online-ddl-0319.jpg)

+ 通过点击 VS Code 中的 `+` 按钮创建一个新终端。

  ![90%](https://public-pingcap-innovation.s3.us-west-2.amazonaws.com/demo-405/405-add-terminal-0319.jpg)

+ 在新终端中运行表结构修改。

  ```bash
  poetry run python mysql_connector_python_demo/online_ddl.py --alter
  ```
  ```output
  Executing ALTER TABLE on online_ddl_test
  Connected to TiDB: 2nR9s2Y6oh4rkp7.6z3VQEZ4@gateway01.us-west-2.prod.aws.tidbcloud.com:4000 using database test
  Table online_ddl_test exists, proceeding with ALTER TABLE
  Executing: ALTER TABLE online_ddl_test ADD COLUMN (ed VARCHAR(10) DEFAULT 'N/A')
  ALTER TABLE completed successfully in 0.39 seconds

  Updated table structure:
    id: int(11)
    k: int(11)
    c: char(120)
    pad: char(60)
    ed: varchar(10)
  ```

  ![90%](https://public-pingcap-innovation.s3.us-west-2.amazonaws.com/demo-405/405-alter-table-0319.jpg)

+ 在 VS Code 中切回第一个终端，观察插入作业将继续运行并显示重试日志。

  ```output
  ...
  Inserting record 105
  Inserting record 106
  Schema mutation encountered, retrying...
  Inserting record 107
  Inserting record 108
  ...
  ```

  ![90%](https://public-pingcap-innovation.s3.us-west-2.amazonaws.com/demo-405/405-error-catch.jpg)

+ 插入 DML 和 DDL 操作都不会被阻塞。

## 模块 6：并发控制 - 悲观模式与乐观模式

在任何支持并发访问的数据库中，都存在冲突的风险，即多个事务可能同时尝试更新相同的数据。虽然设计良好的数据模型可以最小化此类冲突的可能性，但它们仍然可能发生，必须有效管理。数据库提供了处理并发数据访问的机制，主要有悲观和乐观两种并发控制方法。TiDB 支持这两种方法。

本模块展示了 TiDB 中悲观锁和乐观锁的行为。

### 步骤 1. 阅读悲观模式演示代码

+ 在 VS Code 左侧的 **EXPLORER** 中，在 `mysql_connector_python_demo` 文件夹下，点击 `lock_pessimistic.py` 文件。让我们来简单读一下关键代码。

+ 在以下代码中，启动了两个线程。每个线程都将创建一个到 TiDB 的连接。

  ```output
  ...
  t1 = threading.Thread(target=thread1_task)
  t2 = threading.Thread(target=thread2_task)
  t1.start()
  t2.start()
  ...
  ```

+ 在以下代码中，使用 `threading.Event` 来协调线程 1 和线程 2 的动作，以确保冲突一定会发生。线程 1 和 2 都禁用了自动提交。

  ```output
  def thread1_task():
    conn1 = get_connection(autocommit=False)
    cur1 = conn1.cursor(dictionary=True)
    cur1.execute("BEGIN PESSIMISTIC;")
    print("[Thread1] BEGIN PESSIMISTIC", flush=True)

    # notify thread2 to begin transaction
    thread1_begin.set()

    # update and hold lock
    cur1.execute("UPDATE users SET name = 'conn1' WHERE id = 1;")
    print("[Thread1] UPDATED row → name='conn1'", flush=True)

    # notify thread2 to begin transaction
    thread1_updated.set()

    # wait thread2 to attempt update (blocked)
    thread2_ready.wait()

    cur1.execute("COMMIT;")
    print("[Thread1] COMMITTED", flush=True)
    ...
  ```
  ```output
  def thread2_task():
    conn2 = get_connection(autocommit=False)
    cur2 = conn2.cursor(dictionary=True)

    # wait thread1 to complete BEGIN
    thread1_begin.wait()

    cur2.execute("BEGIN PESSIMISTIC;")
    print("[Thread2] BEGIN PESSIMISTIC", flush=True)

    # wait thread1 to update and hold lock
    thread1_updated.wait()

    # notify thread1 that I will attempt UPDATE
    thread2_ready.set()
    print("[Thread2] attempting UPDATE (will block)...", flush=True)

    # this line will be blocked until thread1 COMMIT release row lock
    cur2.execute("UPDATE users SET name = 'conn2' WHERE id = 1;")
    print("[Thread2] UPDATED row → name='conn2'", flush=True)

    cur2.execute("COMMIT;")
    print("[Thread2] COMMITTED", flush=True)
    ...
  ```

### 步骤 2. 在悲观模式下运行演示

+ 在 VS Code 左侧的 **EXPLORER** 中，在 `mysql_connector_python_demo` 文件夹下，点击 `lock_pessimistic.py` 文件。请阅读源代码。

+ 在悲观模式下运行演示代码并观察输出。注意线程二在线程 1 提交之前无法提交。

  ```bash
  poetry run python mysql_connector_python_demo/lock_pessimistic.py
  ```
  ```output
  Create database lab_demo_405_3450001
  [Thread1] BEGIN PESSIMISTIC
  [Thread2] BEGIN PESSIMISTIC
  [Thread1] UPDATED row → name='conn1'
  [Thread2] attempting UPDATE (will block)...
  [Thread1] COMMITTED
  [Thread2] UPDATED row → name='conn2'
  [Thread2] COMMITTED
  Update Result: name is changed to conn2
  ```

### 步骤 3. 阅读乐观模式演示代码

+ 在 VS Code 左侧的 **EXPLORER** 中，在 `mysql_connector_python_demo` 文件夹下，点击 `lock_optimistic.py` 文件。请阅读源代码。

+ 乐观锁的演示代码与悲观锁的类似，不同之处在于两个线程都实现了 `try` 和 `except` 逻辑。此外，乐观锁的演示代码中没有使用 `threading.Event`。

+ 我们以线程 1 为例。

  ```output
  def thread1_task():
    ...
    cur1.execute("BEGIN OPTIMISTIC;")
    print("[Thread1] BEGIN OPTIMISTIC", flush=True)
    
    # update
    cur1.execute("UPDATE users SET name = 'conn1' WHERE id = 1;")
    print("[Thread1] UPDATED row → name='conn1'", flush=True)
    ...

    # thread1 try to commit.
    # error(s) will be raised if there is a conflict
    try:
        cur1.execute("UPDATE users SET name = 'conn1' WHERE id = 1;")
        print("[Thread1] UPDATED row → name='conn1'", flush=True)
        print("[Thread1] Try to commit", flush=True)

        cur1.execute("COMMIT;")
        print("[Thread1] Commit Successful", flush=True)

    except mysql.connector.Error as err:
        print("[Thread1] Conflict occurs: {}".format(err), flush=True)
        print("[Thread1] Retry", flush=True)

        try:
            cur1.execute("BEGIN OPTIMISTIC;")
            print("[Thread1 retry] BEGIN OPTIMISTIC", flush=True)

            cur1.execute("UPDATE users SET name = 'conn1' WHERE id = 1;")
            print("[Thread1 retry] UPDATED row → name='conn1'", flush=True)

            cur1.execute("COMMIT;")
            print("[Thread1 retry] Commit successful", flush=True)

        except mysql.connector.Error as err:
            print("[Thread1 retry] Conflict occurs: {}".format(err), flush=True)
  ```

### 步骤 4. 在乐观模式下运行演示

+ 您可以使用以下命令运行乐观锁的演示代码。由于我们没有使用 `threading.Event` 来控制执行顺序，所以有许多可能的输出。我们在这里列出三个典型输出。在第一个示例输出中，`线程 2` 发生冲突并重试。在第二个示例输出中，`线程 1` 发生冲突。在第三个示例输出中，没有发生冲突。

+ 但是，在这种情况下不会发生事务锁等待。如果发生冲突，解决方案由您自行决定。在我们的演示中，将执行简单的重试。
  
  ```bash
  poetry run python mysql_connector_python_demo/lock_optimistic.py
  ``` 

  ```output
  [Thread1] BEGIN OPTIMISTIC
  [Thread1] UPDATED row → name='conn1'
  [Thread1] UPDATED row → name='conn1'
  [Thread1] Try to commit
  [Thread2] BEGIN OPTIMISTIC
  [Thread1] Commit Successful
  [Thread2] UPDATED row → name='conn2'
  [Thread2] Try to commit
  [Thread2] Conflict occurs: 9007 (HY000): Write conflict, txnStartTS=458156067333603352, conflictStartTS=458156067333603338, conflictCommitTS=458156067333603359, key=????, reason=Optimistic [try again later]
  [Thread2] Retry
  [Thread2 retry] BEGIN OPTIMISTIC
  [Thread2 retry] UPDATED row → name='conn2'
  [Thread2 retry] Commit successful
  Update Result: name is changed to conn2
  ```

  ```output
  [Thread2] BEGIN OPTIMISTIC
  [Thread2] UPDATED row → name='conn2'
  [Thread2] Try to commit
  [Thread1] BEGIN OPTIMISTIC
  [Thread2] Commit Successful
  [Thread1] UPDATED row → name='conn1'
  [Thread1] UPDATED row → name='conn1'
  [Thread1] Try to commit
  [Thread1] Conflict occurs: 9007 (HY000): Write conflict, txnStartTS=458156066350563402, conflictStartTS=458156066350563397, conflictCommitTS=458156066350563406, key=????, reason=Optimistic [try again later]
  [Thread1] Retry
  [Thread1 retry] BEGIN OPTIMISTIC
  [Thread1 retry] UPDATED row → name='conn1'
  [Thread1 retry] Commit successful
  Update Result: name is changed to conn1
  ```

  ```output
  [Thread2] BEGIN OPTIMISTIC
  [Thread2] UPDATED row → name='conn2'
  [Thread2] Try to commit
  [Thread2] Commit Successful
  [Thread1] BEGIN OPTIMISTIC
  [Thread1] UPDATED row → name='conn1'
  [Thread1] UPDATED row → name='conn1'
  [Thread1] Try to commit
  [Thread1] Commit Successful
  Update Result: name is changed to conn1
  ```
  
### 步骤 5. 前往 TiDB Cloud 控制台开始您的旅程

+ 动手实验到此结束。您现在可以在[这里](https://tidbcloud.com/clusters)开始您的 TiDB Cloud 之旅。如果您在本实验中使用了自己的免费层集群，您会在同一页面上找到它，集群名称为 ${LAB:SERVERLESS_CLUSTER_NAME}。

## 总结

在本实验中，您学习了如何使用 Python 代码连接到 TiDB 集群。您还学习了如何使用 `mysql-connector-python` 库与 TiDB 交互，包括连接、插入、单行查询和多行查询。

然后，我们学习了如何使用预处理语句和批量插入来提高应用程序的性能。

之后，我们学习了如何使用 `AUTO_RANDOM` 来避免热点，以及如何在不中断业务的情况下执行在线表结构变更。

最后，我们向您展示了 TiDB 中用于并发控制的悲观锁和乐观锁。