仅将不存在的数据从 CSV 导入数据库答案

【问题标题】：Import only non-existing data to database from CSV仅将不存在的数据从 CSV 导入数据库
【发布时间】：2018-01-27 06:55:09
【问题描述】：

我创建了一个从 CSV 文件读取数据的脚本，检查数据是否已存在于数据库中，如果不存在则将其导入。如果数据确实存在（特定产品的代码），则需要从 CSV 文件更新其余信息。

例如；我的 CSV 文件中有一个代码 WTW-2LT 的成员，名叫 Alex，姓 Johnson。该脚本检查代码 WTW-2LT、名为 Alex 和姓 Johnson 的成员是否已经存在，如果存在，则需要从脚本中更新联系方式和额外详细信息（还需要检查主题和讲师等其他详细信息，所有详细信息都在 CSV 中的一行中），如果它不存在，则只需创建新成员。

我的脚本是我目前所拥有的，目前只进行最少的其他检查以防止分心；

while ($row = fgetcsv($fp, null, ";")) {
    if ($header === null) {
        $header = $row;
        continue;
    }

    $record = array_combine($header, $row);

    $member = $this->em->getRepository(Member::class)->findOneBy([
        'code' =>$record['member_code'],
        'name' =>$record['name'],
        'surname' =>$record['surname'],
    ]);

    if (!$member) {
        $member = new Member();
        $member->setCode($record['member_code']);
        $member->setName($record['name']);
        $member->setName($record['surname']);
    }    
    $member->setContactNumber($record['phone']);
    $member->setAddress($record['address']);
    $member->setEmail($record['email']);

    $subject = $this->em->getRepository(Subject::class)->findOneBy([
        'subject_code' => $record['subj_code']
    ]);

    if (!$subject) {
        $subject = new Subject();
        $subject->setCode($record['subj_code']);
    }
    $subject->setTitle($record['subj_title']);
    $subject->setDescription($record['subj_desc']);
    $subject->setLocation($record['subj_loc']);

    $lecturer = $this->em->getRepository(Lecturer::class)->findOneBy([
        'subject' => $subject,
        'name' => $record['lec_name'],
        'code' => $record['lec_code'],
    ]);

    if (!$lecturer) {
        $lecturer = new Lecturer();
        $lecturer->setSubject($subject);
        $lecturer->setName($record['lec_name']);
        $lecturer->setCode($record['lec_code']);
    }
    $lecturer->setEmail($record['lec_email']);
    $lecturer->setContactNumber($record['lec_phone']);

    $member->setLecturer($lecturer);

    $validationErrors = $this->validator->validate($member);
    if (!count($validationErrors)) {
        $this->em->persist($member);
        $this->em->flush();
    } else {
        // ...
    }
}

您可以注意到此脚本必须查询数据库 3 次以检查是否存在 CSV 行。在我的例子中，我有多达 2000 多行的文件，因此每行执行 3 次查询以检查该行是否存在非常耗时。

不幸的是，我也无法批量导入行，因为如果一个主题不存在，它将创建很多次，直到该批次刷新到数据库，然后我坐在没有任何意义的重复记录中。

如何最大限度地提高性能和速度？就像首先从数据库中获取所有记录并将其存储在数组中（内存消耗？）然后进行检查并将行添加到数组并从那里检查......

有人可以帮我找到改进的方法吗（请提供示例代码？）

【问题讨论】：

标签： php csv doctrine-orm symfony-2.8

【解决方案1】：

老实说，我确实发现 2000 多行的查询数量是该数量的 3 倍，但并没有那么多。但是，由于您要求性能，这是我的两分钱：

使用框架总是会产生开销。这意味着如果您使用本机 PHP 编写此代码，它会运行得更快。我不熟悉 symfony，但我假设您将数据存储在数据库中。在 MySQL 中，您可以使用命令INSERT ... ON DUPLICATE KEY update。如果您已将 3 个字段（代码、姓名、姓氏）设置为主键（我假设），您可以使用它来：插入数据，但如果键已经存在，则更新数据库中的值。 MySQL 会为你做检查，看看数据是否发生了变化：如果没有，就不会发生磁盘写入。

我很确定你可以将原生 SQL 编写到 symfony，允许你使用框架提供的安全性，同时加快你的插入速度。

【讨论】：

嗯，好吧，我想我只需要习惯使用set_time_limit，因为超时主要是问题所在。
如果您不想更改时间限制，您可以制作一个脚本，将这些行转换为单个 SQL 查询。将查询保存为 txt 文件并将其上传到服务器。使用 php，您可以从文件中读取 X 行，在数据库上运行它们并保存最后运行的行。比下一次从那条线继续。然而，这样做需要做更多的工作，但它会更多地平衡服务器负载。

【解决方案2】：

一般来说，如果您想要性能，我最好的经验是将所有数据转储到数据库中，然后使用 SQL 语句在其中进行转换。 DBS 将能够以这种方式优化您的所有步骤。

您可以使用 SQL 命令将 CSV 文件直接导入 MySQL 数据库

LOAD DATA INFILE 'data.csv'
INTO TABLE tmp_import

该命令有很多选项，您可以在其中指定 CSV 文件的格式，例如：

如果您的 data.csv 是一个包含所有旧行和新行的完整转储，那么您可以在稍微修复后将当前表替换为导入的表。

例如，您的 csv 文件（和导入表）可能看起来有点像

WTW-2LT, Alex, Johnson, subj_code1, ..., lec_name1, ...
WTW-2LT, Alex, Johnson, subj_code1, ..., lec_name2, ...
WTW-2LT, Alex, Johnson, subj_code2, ..., lec_name3, ...
WTW-2LU, John, Doe,     subj_code3, ..., lec_name4, ...

然后您可以通过分组获得不同的行：

SELECT member_code, name, surname
FROM tmp_import
GROUP BY member_code, name, surname

如果 member_code 是一个键，你可以在 MySQL 中使用 GROUP BY member_code。即使我相信它在技术上是against the standard，星展银行也不会抱怨。

要获取其余数据，您也可以这样做：

SELECT subj_code, subj_title, member_code
FROM tmp_import
GROUP BY subj_code

和

SELECT lec_code, lec_name, subj_code
FROM tmp_import
GROUP BY lec_code

假设 subj_code 和 lec_code 都是主题和讲座的键。

要实际将此结果保存为表，您可以使用 MySQL 的 CREATE TABLE ... SELECT-syntax，例如

CREATE TABLE tmp_import_members
SELECT member_code, name, surname
FROM tmp_import
GROUP BY member_code, name, surname

然后您可以在两个查询中执行inserts 和updates：

INSERT INTO members (member_code, name, surname)
SELECT member_code, name, surname
FROM tmp_import_members
WHERE tmp_import_members.member_code NOT IN (
  SELECT member_code FROM members WHERE member_code IS NOT NULL
);

UPDATE members 
JOIN tmp_import_members ON 
  members.member_code = tmp_import_members.members_code
SET 
  members.name = tmp_import_members.name,
  members.surname = tmp_import_members.surname;

同样适用于您喜欢的科目和讲座。

这都相当于

一次批量导入 CSV 文件，应该非常快，
为您的成员、主题和讲座提供 3 个临时表，
3 个插入和 3 个更新语句（每个表一个）
完成后将表放在临时表上

再次重申：如果您的 CSV 文件包含所有行，您可以替换现有表并保存 3 个插入和 3 个更新。

确保在临时表的相关列上创建索引，以便 MySQL 可以加快上述查询中的 NOT IN 和 JOIN。

【讨论】：

【解决方案3】：

您可以先运行自定义 sql 以从所有三个表中获取计数

SELECT
  (SELECT COUNT(*) FROM member WHERE someCondition) as memberCount, 
  (SELECT COUNT(*) FROM subject WHERE someCondition) as subjectCount,
  (SELECT COUNT(*) FROM lecturer WHERE someCondition) as lecturerCount

然后根据计数，您可以找到表中是否存在数据。如果您使用本机 SQL，则不必多次运行查询以获得唯一性

查看此链接以了解如何在 Doctrine 中运行自定义 SQL

Symfony2 & Doctrine: Create custom SQL-Query

【讨论】：