从巨大文件中读取和绘制数据

Question

我们有很大的文件，1-1.5GB的顺序（主要是日志文件）与原始数据很容易解析为csv，随后应该生成图形一组图形图像。

目前，我们使用bash脚本将原始数据转换为csv文件，只需要绘制数字，然后将其提供给gnuplot脚本。但是这个过程非常缓慢。我试图通过用单个awk命令替换一些管道cut s，tr s等来加速bash脚本，虽然这提高了速度，但整个过程仍然非常缓慢。

因此，我开始相信这个过程有更好的工具。我目前正在用python + numpy或R重写这个过程。我的一位朋友建议使用JVM，如果我要这样做，我会使用clojure，但不知道JVM如何执行。

我在处理这类问题方面没有太多经验，所以关于如何进行的任何建议都会很棒。谢谢。另外，我将要存储（到磁盘）生成的中间数据，即csv，所以我不必重新生成它，我应该选择我想要一个不同的图表。

编辑2：的原始数据文件具有每一条线，其字段由分隔符（|）是分开的一个记录。并非所有字段都是数字。我在输出csv中需要的每个字段都是通过在输入记录上应用某个公式获得的，这些公式可能会使用输入数据中的多个字段。输出csv每行有3-4个字段，我需要在（可能是）条形图中绘制1-2,1-3,1-4字段的图表。我希望能给出更好的图片。

编辑3：我修改了@ adirau的脚本，它似乎工作得很好。我已经走得够远了，我正在读取数据，发送到处理器线程池（伪处理，将线程名称附加到数据），并通过另一个收集器线程将它聚合到输出文件中。

PS：我不确定这个问题的标记，请随时纠正它。

Answer 1

python的声音是一个不错的选择，因为它有一个很好的线程API（实现是有问题的），matplotlib和pylab。我错过了一些更多的规格，但也许这可能是一个很好的起点：matplotlib: async plotting with threads。 我会去单线程处理大容量磁盘I/O读取和同步队列到线程池进行数据处理（如果您有固定的记录长度，事前可能会通过预计算读取偏移量并将偏移量仅传递到线程池）;使用diskio线程，我会映射数据源文件，读取预定义的num个字节并再读一次，最终将最后一个字节抓到当前数据源行输入的末尾;应该选择接近平均线路输入长度的数字;接下来是通过队列进行游泳池馈送以及线程池中发生的数据处理/绘图;我在这里没有一个好的照片（你准确地绘制了什么），但我希望这有助于。

编辑：有文件。readlines（[sizehint]）一次抓取多行;以及它可能不那么迅速的Cuz文档都在说使用它的ReadLine（）在内部

编辑：快速骨架代码

import threading 
from collections import deque 
import sys 
import mmap 


class processor(Thread): 
    """ 
     processor gets a batch of data at time from the diskio thread 
    """ 
    def __init__(self,q): 
     Thread.__init__(self,name="plotter") 
     self._queue = q 
    def run(self): 
     #get batched data 
     while True: 
      #we wait for a batch 
      dataloop = self.feed(self._queue.get()) 
      try: 
       while True: 
        self.plot(dataloop.next()) 
      except StopIteration: 
       pass 
      #sanitizer exceptions following, maybe 

    def parseline(self,line): 
     """ return a data struct ready for plotting """ 
     raise NotImplementedError 

    def feed(self,databuf): 
     #we yield one-at-time datastruct ready-to-go for plotting 
     for line in databuf: 
      yield self.parseline(line) 

    def plot(self,data): 
     """integrate 
     https://www.esclab.tw/wiki/index.php/Matplotlib#Asynchronous_plotting_with_threads 
     maybe 
     """ 
class sharedq(object): 
    """i dont recall where i got this implementation from 
    you may write a better one""" 
    def __init__(self,maxsize=8192): 
     self.queue = deque() 
     self.barrier = threading.RLock() 
     self.read_c = threading.Condition(self.barrier) 
     self.write_c = threading.Condition(self.barrier) 
     self.msz = maxsize 
    def put(self,item): 
     self.barrier.acquire() 
     while len(self.queue) >= self.msz: 
      self.write_c.wait() 
     self.queue.append(item) 
     self.read_c.notify() 
     self.barrier.release() 
    def get(self): 
     self.barrier.acquire() 
     while not self.queue: 
      self.read_c.wait() 
     item = self.queue.popleft() 
     self.write_c.notify() 
     self.barrier.release() 
     return item 



q = sharedq() 
#sizehint for readine lines 
numbytes=1024 
for i in xrange(8): 
    p = processor(q) 
    p.start() 
for fn in sys.argv[1:] 
    with open(fn, "r+b") as f: 
     #you may want a better sizehint here 
     map = mmap.mmap(f.fileno(), 0) 
     #insert a loop here, i forgot 
     q.put(map.readlines(numbytes)) 

#some cleanup code may be desirable 

Answer 2

我认为Python + numpy的将是最有效的方式，对于速度和易于实施。 Numpy是高度优化的，所以性能很好，python会缓解算法实现部分。

这个组合应该适合您的情况，为您优化内存中的文件加载，尝试找到处理数据块之间的中间点，该数据块不是太大但足够大以最小化读写因为这是什么会减慢程序的速度

如果你觉得这需要更多的加速（我真诚怀疑），你可以使用Cython来加速缓慢的部分。