在Python中实现类似shell的管道性能

Question

[编辑:首先阅读接受的答案.下面的长期调查源于时间测量中的微妙错误.

我经常需要处理包含高度冗余数据的超大(100GB +)文本/类CSV文件,这些文件实际上无法存储在未压缩的磁盘上.我非常依赖外部压缩器,如lz4和zstd,它们产生的stdout流接近1GB/s.

因此,我非常关心Unix shell管道的性能.但是大型shell脚本很难维护,因此我倾向于使用Python构建管道,将命令拼接在一起并仔细使用shlex.quote().

这个过程繁琐且容易出错,因此我想要一种"Pythonic"方式来实现同样的目的,在不卸载的情况下管理Python中的stdin/stdout文件描述符/bin/sh.但是,我从未找到过这样做的方法而不会大大牺牲性能.

Python 3的文档建议使用方法替换shell管道.我已经调整了这个例子来创建下面的测试脚本,它将3GB的数据输入到一个无用的中,它什么都不输出:communicate()subprocess.Popen/dev/zerogrep

#!/usr/bin/env python3
from shlex import quote
from subprocess import Popen, PIPE
from time import perf_counter

BYTE_COUNT = 3_000_000_000
UNQUOTED_HEAD_CMD = ["head", "-c", str(BYTE_COUNT), "/dev/zero"]
UNQUOTED_GREP_CMD = ["grep", "Arbitrary string which will not be found."]

QUOTED_SHELL_PIPELINE = " | ".join(
    " ".join(quote(s) for s in cmd)
    for cmd in [UNQUOTED_HEAD_CMD, UNQUOTED_GREP_CMD]
)

perf_counter()
proc = Popen(QUOTED_SHELL_PIPELINE, shell=True)
proc.wait()
print(f"Time to run using shell pipeline: {perf_counter()} seconds")

perf_counter()
p1 = Popen(UNQUOTED_HEAD_CMD, stdout=PIPE)
p2 = Popen(UNQUOTED_GREP_CMD, stdin=p1.stdout, stdout=PIPE)
p1.stdout.close()
p2.communicate()
print(f"Time to run using subprocess.PIPE: {perf_counter()} seconds")

输出:

Time to run using shell pipeline: 2.412427189 seconds
Time to run using subprocess.PIPE: 4.862174164 seconds

这种subprocess.PIPE方法的速度是速度的两倍多/bin/sh.如果我们将输入大小提高到90GB(BYTE_COUNT = 90_000_000_000),我们确认这不是一个恒定时间开销:

Time to run using shell pipeline: 88.796322932 seconds
Time to run using subprocess.PIPE: 183.734968687 seconds

我到目前为止的假设是,subprocess.PIPE它只是用于连接文件描述符的高级抽象,并且数据永远不会复制到Python进程本身.正如预期的那样,运行上述测试时head使用100%CPU但subproc_test.py使用接近零的CPU和RAM.

鉴于此,为什么我的管道这么慢？这是Python的内在限制subprocess吗？如果是这样的话,/bin/sh在引擎盖下做什么有什么不同呢？

更一般地说,有没有更好的方法在Python中构建大型,高性能的子流程管道？

Answer 1

你时间安排错了。您的perf_counter()通话不会启动和停止计时器；它们只是返回自某个任意起点以来的秒数。该起点可能恰好是perf_counter()这里的第一个呼叫，但它可能是任何点，甚至是将来的一个。

该方法实际花费的时间subprocess.PIPE是 4.862174164 - 2.412427189 = 2.449746975 秒，而不是 4.862174164 秒。此时间并没有显示出可测量的性能损失subprocess.PIPE。