修改数据库参数、CSV文件位置、表名称即可在mysql数据库导入CSV文件，亲测数百万条数据导入可用。需要注意CSV文件编码必须是UTF-8，可以用记事本另存修改。

        此代码使用了批量插入，每次插入1000条数据，避免数据量过大导致无法执行。

以下是源代码：

import pandas as pd
import mysql.connector
from mysql.connector import Error
import os
import time
import numpy as np

def import_csv_to_mysql(csv_file, table_name, db_config):
    """
    将CSV文件导入到MySQL数据库

    参数:
    csv_file: CSV文件路径
    table_name: 数据库表名
    db_config: 数据库连接配置字典
    """
    try:
        # 读取CSV文件 - 将所有列都设为字符串类型以避免混合类型警告
        print("开始读取CSV文件...")

        # 方法1：先读取前几行获取列数，然后为所有列指定字符串类型
        df_sample = pd.read_csv(csv_file, nrows=5)
        total_columns = len(df_sample.columns)

        # 创建所有列的dtype映射，全部设为字符串类型
        dtype_dict = {i: str for i in range(total_columns)}

        # 重新读取整个文件，所有列都设为字符串类型
        df = pd.read_csv(csv_file, dtype=dtype_dict, keep_default_na=False, na_values=[''])
        print(f"成功读取CSV文件，共{len(df)}行数据，{len(df.columns)}列")

        # 显示数据列信息
        print(f"CSV文件列名: {list(df.columns)}")
        print("各列数据类型:")
        for i, col in enumerate(df.columns):
            print(f"  第{i}列 '{col}': {df.dtypes.iloc[i]}")

        # 连接MySQL数据库 - 增加超时设置
        db_config_with_timeout = db_config.copy()
        db_config_with_timeout.update({
            'connect_timeout': 300,
            'pool_size': 10
        })

        connection = mysql.connector.connect(**db_config_with_timeout)
        # 设置会话超时时间
        cursor = connection.cursor()
        cursor.execute("SET SESSION wait_timeout = 28800")  # 8小时
        cursor.execute("SET SESSION interactive_timeout = 28800")  # 8小时
        cursor.execute("SET GLOBAL max_allowed_packet=1073741824")  # 1GB

        # 检查表是否存在
        cursor.execute(f"SHOW TABLES LIKE '{table_name}'")
        table_exists = cursor.fetchone() is not None

        if table_exists:
            print(f"表 '{table_name}' 已存在，将追加数据")
            # 获取现有表的列信息
            cursor.execute(f"DESCRIBE {table_name}")
            existing_columns = [column[0] for column in cursor.fetchall()]
            print(f"现有表的列: {existing_columns}")

            # 检查是否有新列需要添加
            new_columns = set(df.columns) - set(existing_columns)
            if new_columns:
                print(f"发现新列需要添加: {new_columns}")
                for column in new_columns:
                    # 所有新列都设为VARCHAR(255)类型
                    alter_query = f"ALTER TABLE {table_name} ADD COLUMN `{column}` VARCHAR(255)"
                    cursor.execute(alter_query)
                    print(f"已添加列: {column}")
        else:
            # 创建新表 - 所有列都设为VARCHAR类型
            print(f"表 '{table_name}' 不存在，将创建新表")
            # 构建CREATE TABLE语句
            columns_def = []
            for i, column in enumerate(df.columns):
                # 所有列都设为VARCHAR类型，避免任何类型转换问题
                # 可以根据需要调整长度
                if len(column) > 50 or any(keyword in column.lower() for keyword in ['content', 'business_scope', 'description', 'remark', '备注', '描述']):
                    column_type = 'VARCHAR(1000)'
                else:
                    column_type = 'VARCHAR(120)'

                columns_def.append(f"`{column}` {column_type}")

            create_table_query = f"""
            CREATE TABLE {table_name} (
                myid INT AUTO_INCREMENT PRIMARY KEY,
                {', '.join(columns_def)}
            )
            """
            cursor.execute(create_table_query)
            print(f"已创建新表: {table_name}")

        # 插入数据 - 分批插入避免超时
        print("开始插入数据...")

        # 构建INSERT语句
        placeholders = ', '.join(['%s'] * len(df.columns))
        columns = ', '.join([f"`{col}`" for col in df.columns])
        insert_query = f"INSERT INTO {table_name} ({columns}) VALUES ({placeholders})"

        # 分批插入数据
        batch_size = 1000  # 每批插入1000行
        total_rows = len(df)
        inserted_rows = 0

        for i in range(0, total_rows, batch_size):
            end_idx = min(i + batch_size, total_rows)
            batch_df = df.iloc[i:end_idx]

            # 将NaN值替换为None（SQL NULL）
            batch_data = []
            for row in batch_df.itertuples(index=False):
                processed_row = []
                for value in row:
                    if pd.isna(value) or value == 'nan' or value == 'NaN' or value == '' or value == 'None' or value == 'NULL':
                        processed_row.append(None)
                    else:
                        # 确保所有值都是字符串，避免类型问题
                        processed_row.append(str(value) if value is not None else None)
                batch_data.append(tuple(processed_row))

            try:
                cursor.executemany(insert_query, batch_data)
                connection.commit()
                inserted_rows += len(batch_data)
                print(f"已插入 {inserted_rows}/{total_rows} 行数据 ({inserted_rows/total_rows*100:.2f}%)")

                # 每插入10000行休息一下，避免服务器压力过大
                if inserted_rows % 10000 == 0:
                    time.sleep(1)

            except Error as e:
                print(f"插入批次 {i}-{end_idx} 时出错: {e}")
                print(f"问题数据样本: {batch_data[0] if batch_data else '无数据'}")
                connection.rollback()

                # 尝试逐行插入以找到具体错误行
                print("尝试逐行插入以定位错误...")
                success_count = 0
                for j, row_data in enumerate(batch_data):
                    try:
                        cursor.execute(insert_query, row_data)
                        success_count += 1
                    except Error as e:
                        print(f"第 {i+j} 行插入失败: {e}")
                        print(f"失败行数据: {row_data}")
                        # 跳过错误行继续
                        continue

                connection.commit()
                inserted_rows += success_count
                print(f"本批次成功插入 {success_count} 行，跳过 {len(batch_data) - success_count} 行")

        print(f"成功插入 {inserted_rows} 行数据到表 '{table_name}'")

    except Error as e:
        print(f"数据库错误: {e}")
    except Exception as e:
        print(f"错误: {e}")
    finally:
        if 'connection' in locals() and connection.is_connected():
            cursor.close()
            connection.close()
            print("数据库连接已关闭")

# 配置信息 - 在这里修改你的配置
def get_config():
    """返回配置信息，方便修改"""
    return {
        'csv_file': 'd:\\import\\test2.csv',  # 你的CSV文件路径
        'table_name': 'stu_table',   # 数据库表名
        'db_config': {
            'host': 'localhost',
            'database': 'jing',  # 你的数据库名
            'user': 'root',          # 你的用户名
            'password': 'password',      # 你的密码
            'charset': 'utf8mb4',
            'connect_timeout': 300
        }
    }

if __name__ == "__main__":
    # 获取配置
    config = get_config()

    # 检查CSV文件是否存在
    if not os.path.exists(config['csv_file']):
        print(f"错误: CSV文件 '{config['csv_file']}' 不存在")
        print("请确保:")
        print("1. CSV文件存在于当前目录")
        print("2. 在get_config()函数中正确设置了csv_file路径")
        print("3. CSV文件包含列标题")
    else:
        # 执行导入
        print(f"开始导入 {config['csv_file']} 到MySQL数据库...")
        start_time = time.time()
        import_csv_to_mysql(
            csv_file=config['csv_file'],
            table_name=config['table_name'],
            db_config=config['db_config']
        )
        end_time = time.time()
        print(f"导入完成，总耗时: {end_time - start_time:.2f} 秒")