pandas df的流数据答案

【问题标题】：Streaming data for pandas dfpandas df的流数据
【发布时间】：2015-12-12 04:28:27
【问题描述】：

我正在尝试模拟使用 pandas 来访问不断变化的文件。

我有一个文件读取 csv 文件，向其中添加一行，然后随机休眠以模拟批量输入。

import pandas as pd
from time import sleep
import random

df2 = pd.DataFrame(data = [['test','trial']], index=None)

while True:
    df = pd.read_csv('data.csv', header=None)
    df.append(df2)
    df.to_csv('data.csv', index=False)
    sleep(random.uniform(0.025,0.3))

第二个文件通过输出数据框的形状来检查数据的变化：

import pandas as pd

while True:
    df = pd.read_csv('data.csv', header=None, names=['Name','DATE'])
    print(df.shape)

问题在于，当我得到正确形状的 DF 时，有时它会输出 (0x2)。

即：

...
(10x2)
(10x2)
...
(10x2)
(0x2)
(11x2)
(11x2)
...

这确实发生在一些，但不会发生在每个形状变化之间（文件添加到数据框）。

知道第一个脚本打开文件添加数据时会发生这种情况，而第二个脚本无法访问它，因此（0x2），这会发生任何数据丢失吗？

我不能直接访问流，只能访问输出文件。或者还有其他可能的解决方案吗？

编辑

这样做的目的是仅加载新数据（我有一个这样做的代码）并“即时”进行分析。一些分析将包括输出/秒、图形（类似于流图）和一些其他数值计算。

最大的问题是我只能访问 csv 文件，并且我需要能够在不丢失或延迟的情况下分析数据。

【问题讨论】：

目标本质上是什么？像看门狗这样检查文件更改的方法可能是更好的方法吗？
你也可以实现一个锁，这样一次只有一个进程可以打开文件，unix有多种方法可以做到这一点stackoverflow.com/questions/29520587/…。一个进程读取另一个进程可能不会丢失任何数据，但如果您使用数据测试更改，您将得到不正确的输出
看门狗似乎是一个有趣的工具，但不是我想要的。我编辑了我的问题以解释更多。
当流媒体尚未完成数据写入时，“阅读器”[您的代码]似乎正在访问 csv 文件。这是一个种族问题。

标签： python pandas streaming

【解决方案1】：

其中一个脚本正在读取文件，而另一个正在尝试写入文件。两个脚本不能同时访问该文件。就像 Padraic Cunningham 在 cmets 中所说，您可以实现一个锁定文件来解决这个问题。

有一个 python 包可以做到这一点，称为 lockfile，文档为 here。

这是您实施了 lockfile 包的第一个脚本：

import pandas as pd
from time import sleep
import random
from lockfile import FileLock

df2 = pd.DataFrame(data = [['test','trial']], index=None)
lock = FileLock('data.lock')

while True:
    with lock:
        df = pd.read_csv('data.csv', header=None)
        df.append(df2)
        df.to_csv('data.csv', index=False)
    sleep(random.uniform(0.025,0.3))

这是您实现了 lockfile 包的第二个脚本：

import pandas as pd
from time import sleep
from lockfile import FileLock

lock = FileLock('data.lock')

while True:
    with lock:
        df = pd.read_csv('data.csv', header=None, names=['Name','DATE'])
    print(df.shape)
    sleep(0.100)

我添加了 100 毫秒的等待，以便可以减慢控制台的输出速度。

这些脚本将在访问“data.csv”文件之前创建一个名为“data.lock”的文件，并在访问“data.csv”文件后删除文件“data.lock”。在任一脚本中，如果“data.lock”存在，脚本将等待“data.lock”文件不再存在。

【讨论】：

这行不通，因为第一个脚本是对不断更新的文件的模拟。实际上，我只能更改/编辑仅读取文件的第二个脚本

【解决方案2】：

您的模拟脚本读取和写入 data.csv 文件。如果一个脚本以只写方式打开文件而另一个以只读方式打开文件，则您可以同时读取和写入。

考虑到这一点，我将用于写入文件的模拟脚本更改为以下内容：

from time import sleep
import random

while(True):
    with open("data.csv", 'a') as fp:
        fp.write(','.join(['0','1']))
        fp.write('\n')
    sleep(0.010)

在 python 中，使用 'a' 打开文件意味着追加为只写。使用 'a+' 将附加读取和写入访问权限。您必须确保写入文件的代码只会以只写方式打开文件，并且正在读取文件的脚本绝不能尝试写入文件。否则，您将需要实施其他解决方案。

现在您应该能够使用您的第二个脚本进行阅读，而不会出现您提到的问题。

【讨论】：