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

我们有相当大的文件,1-1.5 GB的组合(主要是日志文件)与原始数据的顺序,可以很容易地解析为csv,随后应该将其绘制成图形以生成一组图形图像.

目前,我们正在使用bash脚本将原始数据转换为csv文件,只需要绘制数字,然后将其输入到gnuplot脚本中.但这个过程非常缓慢.我试图通过用一个awk命令替换一些管道剪切,trs等来加速bash脚本,虽然这提高了速度,整个事情仍然很慢.

所以,我开始相信这个过程有更好的工具.我目前正在寻找在python numpy或R中重写这个过程.我的一个朋友建议使用JVM,如果我这样做,我将使用clojure,但我不确定JVM将如何执行.

我在处理这类问题方面没有多少经验,所以关于如何进行的任何建议都会很棒.谢谢.

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

编辑2：原始数据文件每行有一条记录,其字段由分隔符(|)分隔.并非所有字段都是数字.输出csv中我需要的每个字段是通过在输入记录上应用某个公式获得的,输入记录可以使用输入数据中的多个字段.输出csv每行将有3-4个字段,我需要在(可能是)条形图中绘制1-2,1-3,1-4字段的图表.我希望能给出更好的画面.

编辑3：我稍微修改了@ adirau的脚本,似乎工作得很好.我已经足够了,我正在读取数据,发送到处理器线程池(伪处理,将线程名称附加到数据),并通过另一个收集器线程将其聚合到输出文件中.

PS：我不确定这个问题的标记,随时纠正它.

解决方法:

python听起来是个不错的选择,因为它有一个很好的线程API(虽然实现有问题),matplotlib和pylab.我想念你的更多规格,但也许这对你来说可能是一个很好的起点：matplotlib: async plotting with threads.
我会选择一个线程来处理批量磁盘i / o读取和同步排队到一个线程池进行数据处理(如果你有固定的记录长度,通过预先计算读取偏移量并将偏移量传递给线程池,事情可能会变得更快) ;使用diskio线程我将mmap数据源文件,读取预定义的num个字节再读一次,最终获取当前数据源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

标签：python,r,large-files,numpy,graphing
来源： https://codeday.me/bug/20190521/1148869.html