如何在python中比较两个源数据

Question

我有来自 2 个来源的大型数据集，一个是巨大的文本文件（作为新数据），另一个来自数据库 (MySQL)（作为历史数据）。如何比较它们以找出差异（仅获取新的和修改的记录）。我打算读取文本文件中的每一行并与 MySQL 中的数据进行比较。

文本文件（demo.txt）：

0001|NAME1|A1
0002|NAME2|A2
0003|NAME3|A3
0004|NAME4|A4
0005|NAME5|A1

MySQL 中的数据：

id   |   name   |   address
----------------------------
0001 |   NAME1  |   A1
----------------------------
0002 |   NAME2  |   A2
----------------------------
0003 |   NAME3  |   A4
----------------------------
0004 |   NAME4  |   A4
----------------------------

预期结果：

0003|NAME3|A4
0005|NAME5|A1

这是我的努力：

connection = pymysql.connect(host='localhost',
                             user='root',
                             password='password',
                             db='test',
                             charset='utf8',
                             cursorclass=pymysql.cursors.DictCursor)
cursor = connection.cursor()
myfile = open("demo.txt","r")
lines = myfile.readlines()
for line in lines:
    data=line.split('|')
    sql_query = """SELECT * FROM HistoricalTable WHERE id = {}""".format(data[0])
    check=cursor.execute(sql_query)
    result=cursor.fetchall()

    if check>=1:
        for i in result:
            if data[0]==i['id'] and data[1]==i['name'] and data[2]==i['address']:
                print("Ignore")
            else:
                print(line)
                break
    else:
        print(line)

而且我知道这不是最好的方法，而且会花费很多时间！那么请问我有什么建议吗？

Answer 1

您的问题

我可以确定两个问题： - 优化问题， - 关于数据集比较的技术问题。

关于优化

最好有一个工作程序，而不是一个不能完美运行的优化程序。 因此，您可以专注于开发一些功能性的东西，然后对其进行优化。

无论如何，如果您想在两个数据集中查找重复项，则至少需要一个数据集在内存中，除非您的数据集具有一些允许进一步优化的内在属性（例如，我想到了排序）。

关于查找重复项

例如，较小的数据集可以加载到 iterable 中（例如 list），第二个可以使用生成器函数从文件/数据库中迭代。 p>

这是一个演示：

# data1 in memory
data1 = ['a', 'b', 'd', 'c']


# data2 produced by a generator
def item_data2():
    for item in ['a', 'd', 'e', 'c']:
        yield item


duplicates = [x for x in item_data2() if x in data1]
# -> ['a', 'd', 'c']

一些最佳实践

读取 CSV 文件

要读取 CSV 文件，您可以使用 csv 库。在您的具体情况下，您可以使用DictReader。

import csv
import io
import os
import pprint

delimiter = '|'
header = "id|name|address".split(delimiter)

data_dir = "path/to/data/directory"

csv_path = os.path.join(data_dir, "source1.csv")

with io.open(csv_path, mode="r", encoding="utf-8") as f:
    reader = csv.DictReader(f, header, delimiter=delimiter)
    for entry in reader:
        pprint.pprint(entry)

# ->
# OrderedDict([('id', '0001'), ('name', 'NAME1'), ('address', 'A1')])
# OrderedDict([('id', '0002'), ('name', 'NAME2'), ('address', 'A2')])
# OrderedDict([('id', '0003'), ('name', 'NAME3'), ('address', 'A3')])
# OrderedDict([('id', '0004'), ('name', 'NAME4'), ('address', 'A4')])
# OrderedDict([('id', '0005'), ('name', 'NAME5'), ('address', 'A1')])

按id索引条目

如果要比较具有相同 id 的数据，可以将数据存储在关联 id 和其他文件的映射中：

data_mapping = {}
with io.open(csv_path, mode="r", encoding="utf-8") as f:
    reader = csv.DictReader(f, header, delimiter=delimiter)
    for entry in reader:
        data_id = entry.pop('id')
        data_mapping[data_id] = entry

查询数据库

对于数据库，您可以遍历每条记录而不是获取所有记录：您可以改用fetchone。但是，实际上，您可以将光标用作迭代器。

这是一个演示：

import pymysql


connection = pymysql.connect(host='localhost',
                             user='root',
                             password='password',
                             db='test',
                             charset='utf8',
                             cursorclass=pymysql.cursors.DictCursor)

try:
    with connection.cursor() as cursor:
        # Using the cursor as iterator
        cursor.execute("SELECT * FROM HistoricalTable")
        for row in cursor:
            print(row)
finally:
    connection.close()