如何实现从Kafka到MSSQL的高效数据加载？

已尝试的三种加载方法 #

1. 单条插入法 #

速度：每分钟仅500行

while max_offset < end_offset-1:
    msg = consumer.poll() 
    max_offset += 1
if msg is not None:
    if msg.error():
        print(f'Error while receiving message: {msg.error()}')
    else:
        value = msg.value()
        offset = msg.offset()
        try: 
            cursor = conn_str.cursor()
            df = pd.json_normalize(value)
            for index, row in df.iterrows():
                cursor.execute("INSERT INTO TABLE_NAME (COLUMN_1, COLUMN_2, COLUMN_3, COLUMN_4, COLUMN_5, COLUMN_6) values(?,?,?,?,?,?)", row.column_1, row.column_2, row.column_3, row.column_4, row.column_5, row.column_6)
                conn_str.commit()
            
                print(f'Received message:\n{df}')
        except Exception as e:
            print(f'Error message: {e}')
else:
    print('No messages received')

2. 批量插入法 #

速度略有提升

messages_to_insert = []
batch_size = 10000
counter = 0

while max_offset < end_offset-1:
    msg = consumer.poll()
    max_offset += 1
    if msg is not None:
        if msg.error():
           
        else:
         
            row = msg.value()
            offset = msg.offset()
         
            messages_to_insert.append((row["field1"], row["field2"], row["field3"], row["field4"], row["field5"], row["field6"]))
            counter += 1
           
            if counter >= batch_size:
                cursor.executemany("INSERT INTO <table_name> (field1, field2, field3, field4, field5, field6) values(?,?,?,?,?,?)", messages_to_insert)
                conn_str.commit()
                messages_to_insert = []
                counter = 0

if messages_to_insert:
    cursor.executemany("INSERT INTO <table_name> (field1, field2, field3, field4, field5, field6) values(?,?,?,?,?,?)", messages_to_insert)
    conn_str.commit()

3. 先写CSV再导入MSSQL #

写入CSV速度快，但导入MSSQL速度仍慢，SSIS导入也无改善

messages_to_insert = []
batch_size = 10000
counter = 0

with open('data.csv', 'w', newline='') as file:
    writer = csv.writer(file)
    writer.writerow(['COLUMN_1', 'COLUMN_2', 'COLUMN_3', 'COLUMN_4', 'COLUMN_5', 'COLUMN_6'])

    while max_offset < end_offset-1:
        msg = consumer.poll()
        max_offset += 1
        if msg is not None:
            if msg.error():
                logger.error(f'Error while receiving message: {msg.error()}')
            else:
                row = msg.value()
                offset = msg.offset()

                # Write row to file
                writer.writerow([row['field1'], row['field2'], row['field3'], row['field4'], row['field5'], field6])

file_path = 'data.csv'


with open(file_path, 'r') as f:
    lines = f.readlines()
    lines.pop(0)
data = [tuple(line.strip().split(',')) for line in lines]
placeholders = ','.join('?' * len(data[0]))


    query = f"INSERT INTO <table_name> (COLUMN_1, COLUMN_2, COLUMN_3, COLUMN_4, COLUMN_5, COLUMN_6) VALUES ({placeholders})"


    cursor = conn_str.cursor()
    cursor.executemany(query, data)
    cursor.commit()

conn.close()               

高效加载优化方案 #

1. 使用MSSQL原生批量导入工具 #

MSSQL的BULK INSERT或bcp工具专为大数据量设计，性能远高于普通INSERT。可以将Kafka消息写入临时文件后直接调用工具导入：

import subprocess

# 用BULK INSERT执行导入
bulk_query = """
BULK INSERT <table_name>
FROM 'C:\\temp\\kafka_data.csv'
WITH (
    FIELDTERMINATOR = ',',
    ROWTERMINATOR = '\\n',
    FIRSTROW = 2,
    TABLOCK  # 减少锁竞争
)
"""
cursor.execute(bulk_query)
conn_str.commit()

# 或用bcp命令行（性能更优）
bcp_cmd = [
    'bcp', 'your_db.dbo.your_table', 'in', 'kafka_data.csv',
    '-S', 'your_server', '-U', 'user', '-P', 'password',
    '-c', '-t,', '-r\\n', '-F2'
]
subprocess.run(bcp_cmd, check=True)

2. 优化自定义批量插入逻辑 #

针对原批量代码的短板，做以下优化：

• 提前初始化游标，避免重复创建
• 增大批次大小（如50000，根据内存调整）
• 减少提交频率，按批次间隔提交

优化后代码：

cursor = conn_str.cursor()
batch_size = 50000
messages = []
commit_interval = 5  # 每5个批次提交一次

while max_offset < end_offset-1:
    msg = consumer.poll()
    max_offset += 1
    if msg and not msg.error():
        row = msg.value()
        messages.append((row["field1"], row["field2"], row["field3"], row["field4"], row["field5"], row["field6"]))
        
        if len(messages) >= batch_size:
            cursor.executemany("INSERT INTO <table_name> (field1, field2, field3, field4, field5, field6) values(?,?,?,?,?,?)", messages)
            messages = []
            
            # 按间隔提交
            if (max_offset // batch_size) % commit_interval == 0:
                conn_str.commit()

# 提交剩余数据
if messages:
    cursor.executemany("INSERT INTO <table_name> (field1, field2, field3, field4, field5, field6) values(?,?,?,?,?,?)", messages)
conn_str.commit()
cursor.close()

3. 使用Kafka Connect无代码同步 #

Kafka官方提供MSSQL Sink连接器，专为批量同步设计，无需手写消费代码，性能优异。配置示例：

name=mssql-sink
connector.class=io.confluent.connect.jdbc.JdbcSinkConnector
tasks.max=4
topics=your_kafka_topic
connection.url=jdbc:sqlserver://your_server:1433;databaseName=your_db;user=user;password=pass
insert.mode=insert
batch.size=100000
table.name.format=your_table
pk.mode=none

启动连接器后，Kafka会自动将topic数据批量写入MSSQL，无需手动处理消费逻辑。

4. 数据库端性能优化 #

• 插入前禁用目标表的非聚集索引，插入完成后重建，避免索引更新开销
• 开启TABLOCK提示，减少锁竞争
• 关闭自动提交，手动控制提交时机

示例：

# 禁用索引
cursor.execute("ALTER INDEX ALL ON <table_name> DISABLE")

# 执行批量插入逻辑...

# 重建索引
cursor.execute("ALTER INDEX ALL ON <table_name> REBUILD")
conn_str.commit()

内容的提问来源于stack exchange，提问作者Denys