影刀RPA 数据库批量操作：批量插入与更新

影刀RPA 数据库批量操作批量插入与更新署名林焱什么情况用RPA采集了5000条商品数据一条一条INSERT到数据库等到天黑。或者ERP里有3万条客户信息需要更新状态单条UPDATE跑一小时还没完。批量操作是数据库的命脉。executemany一次提交一批速度比单条循环快100倍。但批量操作也有坑——数据格式不对、中间某条失败、数据量太大撑爆内存——都得提前防。典型场景采集结果一次性写入数据库几千到几万条Excel数据批量导入数据库定时同步两个数据库之间的数据批量更新订单状态、客户标签怎么做1. 批量插入基础拼多多店群自动化报活动上架importsqlite3fromdatetimeimportdatetime db_pathrC:\RPAData\orders.dbconnsqlite3.connect(db_path)cursorconn.cursor()# 确保表存在cursor.execute( CREATE TABLE IF NOT EXISTS orders ( id INTEGER PRIMARY KEY AUTOINCREMENT, order_no TEXT UNIQUE, customer TEXT, amount REAL, status TEXT, create_time TEXT ) )# 模拟采集到的订单数据实际从影刀采集组件获取orders[(fORD-{i:05d},f客户{i%100},round(99.9i*0.5,2),pending,datetime.now().strftime(%Y-%m-%d %H:%M:%S))foriinrange(1,5001)# 5000条]# 方式1executemany批量插入cursor.executemany(INSERT OR IGNORE INTO orders VALUES (NULL, ?, ?, ?, ?, ?),orders)conn.commit()print(f插入完成{cursor.rowcount}条)conn.close()2. 分批插入防内存溢出importsqlite3fromdatetimeimportdatetime db_pathrC:\RPAData\orders.dbconnsqlite3.connect(db_path)cursorconn.cursor()# 大量数据分批插入all_data[(fORD-{i:05d},f客户{i%100},round(99.9i*0.5,2),pending,datetime.now().strftime(%Y-%m-%d %H:%M:%S))foriinrange(1,50001)# 5万条]BATCH_SIZE2000total_inserted0foriinrange(0,len(all_data),BATCH_SIZE):batchall_data[i:iBATCH_SIZE]cursor.executemany(INSERT OR IGNORE INTO orders VALUES (NULL, ?, ?, ?, ?, ?),batch)conn.commit()# 每批提交total_insertedlen(batch)print(f进度{total_inserted}/{len(all_data)})print(f全部完成共插入{total_inserted}条)conn.close()影刀操作配合用【循环-列表】组件遍历采集结果每攒满2000条调用一次【执行Python代码】写入。3. 批量更新CASE WHEN方式importsqlite3 db_pathrC:\RPAData\orders.dbconnsqlite3.connect(db_path)cursorconn.cursor()# 需要更新的数据[(新状态, 订单号), ...]updates[(shipped,ORD-00001),(shipped,ORD-00002),(cancelled,ORD-00003),(delivered,ORD-00004),(delivered,ORD-00005),]# 方式1逐条更新慢不推荐# for status, order_no in updates:# cursor.execute(UPDATE orders SET status ? WHERE order_no ?, (status, order_no))# 方式2CASE WHEN批量更新快推荐cursor.execute( UPDATE orders SET status CASE order_no .join([fWHEN ? THEN ?for_inupdates]) ELSE status END WHERE order_no IN (,.join([?for_inupdates])) ,[valforpairin[(s,n)forn,sinupdates]forvalinpair][nforn,sinupdates])conn.commit()print(f更新了{cursor.rowcount}条)conn.close()4. INSERT ON CONFLICT有则更新无则插入importsqlite3fromdatetimeimportdatetime db_pathrC:\RPAData\orders.dbconnsqlite3.connect(db_path)cursorconn.cursor()# 采集数据可能包含已有订单需更新价格和新订单需插入orders[(ORD-00001,客户1,109.9,pending,datetime.now().strftime(%Y-%m-%d %H:%M:%S)),# 已存在更新价格(ORD-00002,客户2,159.9,pending,datetime.now().strftime(%Y-%m-%d %H:%M:%S)),# 已存在更新价格(ORD-50001,新客户,299.0,pending,datetime.now().strftime(%Y-%m-%d %H:%M:%S)),# 新增]# SQLite的UPSERT语法cursor.executemany( INSERT INTO orders (order_no, customer, amount, status, create_time) VALUES (?, ?, ?, ?, ?) ON CONFLICT(order_no) DO UPDATE SET amount excluded.amount, status excluded.status ,orders)conn.commit()print(f插入/更新了{cursor.rowcount}条)conn.close()5. 从Excel批量导入数据库importsqlite3importpandasaspd db_pathrC:\RPAData\orders.dbexcel_pathrC:\RPAData\订单数据.xlsx# 读取Exceldfpd.read_excel(excel_path,dtype{order_no:str,phone:str})# 数据清洗dfdf.dropna(subset[order_no])# 删除订单号为空的行df[amount]pd.to_numeric(df[amount],errorscoerce).fillna(0)# 转为元组列表records[tuple(x)forxindf[[order_no,customer,amount,status]].values]# 批量写入connsqlite3.connect(db_path)cursorconn.cursor()BATCH_SIZE1000foriinrange(0,len(records),BATCH_SIZE):batchrecords[i:iBATCH_SIZE]cursor.executemany( INSERT OR REPLACE INTO orders (order_no, customer, amount, status, create_time) VALUES (?, ?, ?, ?, datetime(now)) ,batch)conn.commit()print(f从Excel导入{len(records)}条到数据库)conn.close()有什么坑坑1executemany中间某条失败前面已执行的不回滚executemany不是原子操作——如果第2000条数据违反约束前1999条已经入库了但你拿到的是异常。解决办法# 方案1用事务推荐try:cursor.executemany(sql,batch)conn.commit()exceptExceptionase:conn.rollback()# 整批回滚print(f本批失败已回滚{e})# 把失败批次写入错误日志跳过继续error_log.append(batch)# 方案2预校验数据defvalidate_records(records):插入前校验valid[]invalid[]forrinrecords:ifnotr[0]:# 订单号为空invalid.append(r)else:valid.append(r)returnvalid,invalid坑2executemany和VALUES多行拼接的性能差异executemany底层是循环执行单条SQL性能比拼接INSERT INTO t VALUES (1,a),(2,b),(3,c)差。但拼接方式有SQL长度限制SQLite默认1000000字节MySQL默认4MB。实际经验数据量500以下用拼接500以上用executemany分批。或者用MySQL的LOAD DATA INFILE速度最快TEMU店群矩阵自动化运营核价报活动# MySQL专用从CSV文件直接导入比INSERT快10倍cursor.execute( LOAD DATA LOCAL INFILE /tmp/orders.csv INTO TABLE orders FIELDS TERMINATED BY , LINES TERMINATED BY \\n  IGNORE 1 ROWS (order_no, customer, amount, status) )坑3自增ID断号用了INSERT OR IGNORE去重被IGNORE的行不会占用ID但SQLite的自增ID还是会跳号。如果你发现ID从100直接跳到500不是bug是IGNORE掉的行消耗了ID号。如果在意ID连续性不要用自增ID做业务编号用业务字段订单号做主键。坑4批量更新时CASE WHEN的SQL太长CASE WHEN方式拼接500条更新SQL语句可能超过1MBMySQL会报错。解决分批拼接。defbatch_update_case_when(cursor,conn,updates,batch_size100):CASE WHEN分批更新foriinrange(0,len(updates),batch_size):batchupdates[i:ibatch_size]case_sql .join([fWHEN ? THEN ?for_inbatch])where_sql,.join([?for_inbatch])params[vforpairin[(s,n)forn,sinbatch]forvinpair][nforn,sinbatch]cursor.execute(f UPDATE orders SET status CASE order_no{case_sql}ELSE status END WHERE order_no IN ({where_sql}) ,params)conn.commit()坑5pandas to_sql的性能问题pandas的df.to_sql()方便但慢5万条数据可能要几分钟。原因是它底层也是逐条INSERT。如果数据量大用原生executemany# 慢pandas to_sqldf.to_sql(orders,conn,if_existsappend,indexFalse)# 快原生executemanyrecordsdf.values.tolist()cursor.executemany(INSERT INTO orders VALUES (?,?,?,?,?,?),records)conn.commit()