我有一个包含大量数据的文件.每行都是一条记录.我试图对整个文件做一些ETL工作.现在我正在使用标准输入逐行读取数据.关于这一点很酷的是你的脚本可以非常灵活地与其他脚本和shell命令集成.我将结果写入标准输出.例如.
$cat input_file
line1
line2
line3
line4
...
我当前的python代码如下所示 – parse.py
import sys
for line in sys.stdin:
result = ETL(line) # ETL is some self defined function which takes a while to execute.
print result
下面的代码是它现在的工作方式：
cat input_file | python parse.py > output_file
我查看了Python的Threading模块,我想知道如果我使用该模块,性能是否会得到显着改善.
问题1：我应该如何规划每个线程的配额,为什么？
...
counter = 0
buffer = []
for line in sys.stdin:
buffer.append(line)
if counter % 5 == 0: # maybe assign 5 rows to each thread? if not, is there a rule of thumb to determine
counter = 0
thread = parser(buffer)
buffer = []
thread.start()
问题2：多线程可能会同时将结果打印回stdout,如何组织它们并避免下面的情况？
import threading
import time
class parser(threading.Thread):
def __init__ (self, data_input):
threading.Thread.__init__(self)
self.data_input = data_input
def run(self):
for elem in self.data_input:
time.sleep(3)
print elem + 'Finished'
work = ['a', 'b', 'c', 'd', 'e', 'f']
thread1 = parser(['a', 'b'])
thread2 = parser(['c', 'd'])
thread3 = parser(['e', 'f'])
thread1.start()
thread2.start()
thread3.start()
输出真的很难看,其中一行包含两个线程的输出.
aFinished
cFinishedeFinished
bFinished
fFinished
dFinished
解决方法:
首先考虑你的第二个问题,这就是mutexes的用途.您可以通过使用锁来协调解析器之间来获得所需的更清晰的输出,并确保在给定的时间段内只有一个线程可以访问输出流：
class parser(threading.Thread):
output_lock = threading.Lock()
def __init__ (self, data_input):
threading.Thread.__init__(self)
self.data_input = data_input
def run(self):
for elem in self.data_input:
time.sleep(3)
with self.output_lock:
print elem + 'Finished'
关于您的第一个问题,请注意,多线程可能无法为您的特定工作负载带来任何好处.这在很大程度上取决于您对每个输入行(您的ETL函数)所做的工作是主要是CPU绑定还是IO绑定.如果是前者(我怀疑可能),由于global interpreter lock,线程将无济于事.在这种情况下,您可能希望使用多处理模块在多个进程之间分配工作而不是多个线程.
但是您可以通过更容易实现的工作流程获得相同的结果：将输入文件拆分为n个(使用例如split命令);在每个子文件上分别调用extract-and-transform脚本;然后连接生成的输出文件.
一个挑剔：“使用标准输入逐行读取数据,因为它不会将整个文件加载到内存中”涉及一个误解.您可以在Python中逐行读取文件,例如,将sys.stdin替换为以下构造中的文件对象：
for line in sys.stdin:
另请参见文件对象的readline()方法,并注意read()可以将要读取的最大字节数作为参数.
标签：python,multithreading,stdin,stdout
来源： https://codeday.me/bug/20190624/1282737.html