读取多个 CSV 文件并使用 python 多处理将数据插入到表中，而不使用 pandas

Question

我在一个文件夹中有 3 个区域特定的 CSV 文件，即 data_cityA.csv、data_cityB.csv 和 data_cityC.csv。我必须阅读并识别特定于区域的文件；将其插入到表中，并添加一个额外的列，该列将包含有关特定区域的信息。

list_of_file=glob.glob('./*csv')
for file_name in list_of_files:
    count = 0
    total = 0   

    with open(file_name,'r')as csvfile:
        read=csv.reader(csvfile)
        next(read)
        if "cityA" in file_name:
            reg="cityA"
        elif "cityB" in file_name:
            reg="cityB"
        elif "cityC" in file_name:
            reg="cityC"

        with open(file_name, 'r')as csv_file:
            reader=csv.reader(csv_file)
            data=list(reader)
            total=len(data)     
            temp_data=[]

        for row in read:
            row.append(reg) #concatenating region name 
            temp_data.append(tuple(row))
            count+=1
            total=-1

            if count>999 or total==1:
                insert_query="INSERT INTO table_name(A,B,C,D,E) values (1,2,3,4,5)"
                curser.executemoany(insert_query,temp_data)
                conn.commit()
                count=0
                insert_query=" "
                temp_data=[]

cursor.callproc('any_proc')
conn.close()

处理大约需要 4-5 小时（数据大小 sybase。有任何想法吗？有没有比多处理更好的方法？

Answer 1

您在这里遇到的一个大问题是：data=list(reader)。这将立即将整个文件读入内存。如果文件为 500MB，则一次将 500MB 加载到内存中。您的另一个选择是使用 reader 作为迭代器。这样做的缺点是您事先不知道记录的总数，因此在退出循环后，您必须执行剩余行的插入。

可能会对您的表现产生重大影响的第二件事是插入。您可以使用多处理并行化它（下面是使用 Pool 的建议），但由于它是一个新进程，您将不得不再次处理连接到数据库（然后关闭它）。

from multiprocessing.pool import Pool
list_of_files = glob.glob('./*csv')
pool = Pool()
pool.map(process_file, list_of_files)
pool.close()
pool.join()
cursor.callproc('any_proc')
conn.close()


def process_file(file_name):
    # Make a new connection
    # conn = ...
    cursor = conn.cursor()
    temp_data = []

    def do_insert():
        insert_query = "INSERT INTO table_name(A,B,C,D,E) values (1,2,3,4,5)"
        cursor.executemany(insert_query, temp_data)
        conn.commit()

    with open(file_name, 'r')as csvfile:
        read = csv.reader(csvfile)
        next(read)
        if "cityA" in file_name:
            reg = "cityA"
        elif "cityB" in file_name:
            reg = "cityB"
        elif "cityC" in file_name:
            reg = "cityC"

        for row in read:
            row.append(reg)  # concatenating region name
            temp_data.append(tuple(row))
            if len(temp_data) > 999:
                do_insert()
                temp_data = []
    if temp_data:
        do_insert()
    conn.close()

Answer 2

数据库往返会拖慢您的速度。

您实际上是每行进行 1 次往返。 500MB 听起来像是很多行......所以这是很多往返。检查 sybase 中是否有一种方法可以提供 csv 并将其加载到表中。更少的调用（甚至可能是 1 个），有很多行，而不是每次调用 1 行。

Answer 3

也许您可以考虑在 Python 之外执行此操作。

考虑下表...

create table t1 ( 
    k int not null, 
    v varchar(255) null, 
    city varchar(255) null)
go

...以及文件“file.txt”

1,Line 1
2,Line 2
3,Line 3
4,Line 4
5,Line 5

注意文件末尾不要有空行。

使用“Stream 编辑器”添加额外的列，在本例中为“CityA”

cat file.txt | sed s/$/\,CityA/g > file_2.txt
cat file_2.txt
1,Line 1,CityA
2,Line 2,CityA
3,Line 3,CityA
4,Line 4,CityA
5,Line 5,CityA

确保数据库已配置为进行批量复制，您的 DBA 可以提供帮助。

use master
go
sp_dboption 'db_name', 'select', true
go

然后使用 Sybase 的 bcp 实用程序加载文件：

bcp database.owner.table in file_2.txt -U login -S server -c -t, -Y -b 1000

参数如下：

• 数据库名称
• 对象所有者
• 表名
• 方向
• 文件名
• -U 用户名
• -S 服务器名（Sybase 实例名而不是物理主机名）
• -c = 使用字符数据
• -t, = 字段终止符是,
• -Y 客户端字符集转换 - 可能不需要
• -b 1000 = 一次提交 1000 行。如果您正在加载 500MB，您可能需要这样做，以免触发 LOG_SUSPEND。