python - 读取和绘制从大文件中读取的数据

Question

我们有相当大的文件，1-1.5 GB 的数量级(主要是日志文件)与可以轻松解析为 csv 的原始数据相结合，随后应该将其绘制成图形以生成一组图形图像。

目前，我们正在使用 bash 脚本将原始数据转换为 csv 文件，其中仅包含需要绘制图形的数字，然后将其输入 gnuplot 脚本。但这个过程极其缓慢。我试图通过用单个 awk 命令替换一些管道 cut、tr 等来加速 bash 脚本，尽管这改进了速度，整个事情还是很慢。

因此，我开始相信这个过程有更好的工具。我目前正在寻找用 python+numpy 或 R 重写这个过程。我的一个 friend 建议使用 JVM，如果我要这样做，我会使用 clojure，但我不确定 JVM 的性能如何。

我在处理这类问题方面没有太多经验，所以任何关于如何进行的建议都会很好。谢谢。

编辑:另外，我想存储(到磁盘)生成的中间数据，即 csv，所以我不必重新生成它，如果我选择我想要一个不同的图表。

编辑2:原始数据文件每行一条记录，其字段由分隔符(|)分隔。并非所有字段都是数字。我在输出 csv 中需要的每个字段都是通过对输入记录应用特定公式获得的，这可能会使用输入数据中的多个字段。输出 csv 每行将有 3-4 个字段，我需要在(可能是)条形图中绘制 1-2、1-3、1-4 个字段的图表。我希望这能提供更好的画面。

编辑 3: 我对@adirau 的脚本做了一些修改，它似乎运行良好。我已经走得足够远了，我正在读取数据，发送到处理器线程池(伪处理，将线程名称附加到数据)，并通过另一个 collector 线程将其聚合到输出文件中。

PS:我对这个问题的标注不太清楚，欢迎指正。

Answer 1

python 听起来是个不错的选择，因为它有一个很好的线程 API(尽管实现是有问题的)、matplotlib 和 pylab。我错过了您的更多规范，但也许这对您来说是一个很好的起点:matplotlib: async plotting with threads .我会选择一个线程来处理批量磁盘 i/o 读取和同步排队到线程池以进行数据处理(如果你有固定的记录长度，通过预先计算读取偏移量并将偏移量传递给线程池，事情可能会变得更快) ;使用 diskio 线程，我将映射数据源文件，读取预定义的 num 个字节 + 再读取一次以最终将最后一个字节抓取到当前数据源 lineinput 的末尾； numbytes 应该选择在你的平均 lineinput 长度附近的某个地方；接下来是通过队列和线程池中发生的数据处理/绘图进行池馈送；我在这里没有很好的图片(你到底在策划什么)但我希望这会有所帮助。

编辑:有 file.readlines([sizehint]) 一次获取多行；好吧，它可能不会那么快，因为文档说它在内部使用 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