Python - CSV：具有不同长度行的大文件

Question

简而言之，我有一个包含2000万行的csv文件，行长度不同。这是由于过时的数据记录器和专有格式导致的。我们得到的结果是一个csv文件，格式如下。我的目标是将此文件插入到Postgres数据库中。我该如何做到以下几点：

• 保留前8列和最后2列，以使CSV文件一致
• 在CSV文件的第一或最后位置添加新列。

---

1, 2, 3, 4, 5, 0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0, img_id.jpg, -50
1, 2, 3, 4, 5, 0,0,0,0,0,0,0,0,0, img_id.jpg, -50
1, 2, 3, 4, 5, 0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0, img_id.jpg, -50
1, 2, 3, 4, 5, 0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0, img_id.jpg, -50
1, 2, 3, 4, 5, 0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0, img_id.jpg, -50
1, 2, 3, 4, 5, 0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0, img_id.jpg, -50
1, 2, 3, 4, 5, 0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0, img_id.jpg, -50
1, 2, 3, 4, 5, 0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0, img_id.jpg, -50
1, 2, 3, 4, 5, 0,0,0,0,0,0, img_id.jpg, -50
1, 2, 3, 4, 5, 0,0,0,0,0,0,0,0,0,0,0 img_id.jpg, -50

- dassouki

---

你能具体说明一下“向文件添加列”是什么意思吗？这一列的数据从哪里获取？ - Mark Byers

告诉我们你遇到了哪一步问题会有所帮助。你不知道如何将行分隔成列吗？你不知道如何跳过某些列吗？如果你只是不知道如何开始，请继续说明，有人也可以回答这个问题。 - Wesley

@Mark Byers，添加列可以在开头或结尾进行。 @Wesley，我卡住的部分是如何跳过某些列；不过，我认为Ignacio可能已经在下面回答了。 - dassouki

Answer 1

使用csv读取一行数据，然后：

newrow = row[:8] + row[-2:]

然后添加新字段并将其写出（同样使用csv）。

- Ignacio Vazquez-Abrams

Answer 2

您可以将文件作为文本文件打开，逐行阅读。如果没有引号或转义逗号来“分割字段”，则可以执行以下操作：

with open('thebigfile.csv', 'r') as thecsv:
    for line in thecsv:
        fields = [f.strip() for f in thecsv.split(',')]
        consist = fields[:8] + fields[-2:] + ['onemore']
        ... use the `consist` list as warranted ...

我猜想在我的代码中的+ ['onemore']，你可能想要“添加一列”并填入一些非常不同的内容，但是我当然无法猜测它可能是什么。

不要将每行作为单独的插入语句发送到数据库--2000万次插入需要很长时间。相反，将“已匹配”的列表分组，将它们附加到一个临时列表上--每当该列表的长度达到1000时，使用executemany来添加所有这些条目。

编辑：澄清一下，我不建议使用csv处理明知不符合“规范”csv格式的文件：直接处理可以给您更直接的控制（特别是当您发现每行逗号数量之外的其他不规则时）。

- Alex Martelli

---

这个回答中有很多好的建议。我不知道为什么会有一个踩。点了+1。 - YOU

嗯..似乎存在一个问题，带引号逗号的行会导致错误。 - Charles Merriam

@Charles，没错，这就是我说 “如果不行，...” 的原因；-）。 如果你需要一个更复杂的“语法”来解析那个“不完美的CSV”并提取字段，你可以逐步应用所需的解析细化（虽然csv模块对于处理_好的csv文件很好，但不能有太多对于输入数据中缺陷的变通方法）。 - Alex Martelli

是的，它确实如此。如果这不是一次性导入，我可能会考虑SnapLogic。你已经放置了所有免责声明，但我怀疑20MLOD不会没有一个引用字符串。 - Charles Merriam

Answer 3

我建议使用csv模块。这里是一些基于CSV处理的代码，我在其他地方做过。

from __future__ import with_statement
import csv

def process( reader, writer):
    for line in reader:
        data = row[:8] + row[-2:]
        writer.write( data )

def main( infilename, outfilename ):
    with open( infilename, 'rU' ) as infile:
        reader = csv.reader( infile )
        with open( outfilename, 'w') as outfile:
            writer = csv.writer( outfile )
            process( reader, writer )

if __name__ == '__main__':
    if len(sys.argv) != 3:
        print "syntax: python process.py filename outname"
        sys.exit(1)
    main( sys.argv[1], sys.argv[2] )

- Seth Johnson

---

出于好奇，你为什么使用了with而不是for？ - dassouki

@dassouki：for 循环在 process() 函数中。 - Ignacio Vazquez-Abrams

Answer 4

抱歉，您需要编写一些代码来解决这个问题。当您有一个像这样的大文件时，值得检查所有内容以确保它与您的期望一致。如果您让不良数据进入数据库，您将永远无法完全清除。

请记住CSV的奇怪之处：它是一堆类似标准的混合物，具有不同的引用、转义、空字符、Unicode、空字段（",,,"）、多行输入和空行规则。csv模块具有“方言”和选项，您可能会发现csv.Sniffer类有所帮助。

我建议您：

• 运行“tail”命令查看最后几行。
• 如果它表现良好，将整个文件通过csv阅读器运行以查看是否有错误。制作一个“每行字段”的快速直方图。
• 考虑“有效”范围和字符类型，并在阅读时严格检查它们。特别注意不寻常的Unicode或可打印范围之外的字符。
• 认真考虑是否要将额外的奇怪值保留在“其余行”文本字段中。
• 将任何意外的行抛到异常文件中。
• 修复您的代码以处理异常文件中的新模式。反复执行。
• 最后，再次运行整个程序，实际将数据转储到数据库中。

在完全完成之前不要触碰数据库，这样可以加快开发时间。此外，请注意SQLite在只读数据上非常快，因此PostGres可能不是最佳解决方案。

您的最终代码可能看起来像这样，但我不确定您的数据，特别是它是否“表现良好”：

while not eof
    out = []
    for chunk in range(1000):
       try:
          fields = csv.reader.next()
       except StopIteration:
          break
       except:
          print str(reader.line_num) + ", 'failed to parse'"
       try:
          assert len(fields) > 5 and len(fields < 12)
          assert int(fields[3]) > 0 and int(fields[3]) < 999999
          assert int(fields[4]) >= 1 and int(fields[4] <= 12) # date
          assert field[5] == field[5].strip()  # no extra whitespace
          assert not field[5].strip(printable_chars)  # no odd chars
          ...
       except AssertionError:
          print str(reader.line_num) + ", 'failed checks'"
       new_rec = [reader.line_num]  # new first item
       new_rec.extend(fields[:8])   # first eight
       new_rec.extend(fields[-2:])  # last two
       new_rec.append(",".join(field[8:-2])) # and the rest
       out.append(new_rec)
    if database:
       cursor.execute_many("INSERT INTO raw_table VALUES %d,...", out)

当然，这段代码的效果因人而异。这只是伪代码的第一稿。预计编写出适用于输入的稳定代码需要大部分时间。

- Charles Merriam