如何使用 Dask 比较两个大型 CSV 文件

Question

我有两个 CSV 文件（每个大约 4GB），我想检查这两个文件的条目之间的差异。

假设 1.csv 中的第 1 行条目与 2.csv 的第 1 行不匹配，但与 2.csv 的第 100 行相同，那么它不应该显示任何差异。

只有在两个 CSV 文件中没有相同的条目时才能看到差异。 约束是不能使用任何数据库。

我正在使用 dask.Dataframe 输入这些文件，但我没有看到任何 api 或函数来查找 Dask 文档中的差异。

我什至还没有找到将 Dask Dataframes 转换为 Panda Dataframes 的方法，我也无法将此 Dataframes 转换为任何文本或 CSV 文件。

有什么解决方案可以比较这么大的文件并找出差异。

请找到我尝试过的示例代码。

import dask.dataframe as dd
import numpy.testing as npt
import pandas as pd

filename1 = '/Users/saikatbhattacharjee/Downloads/2008.csv'
df1 = dd.read_csv(filename1, assume_missing=True)
filename2 = '/Users/saikatbhattacharjee/Downloads/2009.csv'
df2 = dd.read_csv(filename2, assume_missing=True )

def assert_frames_equal(actual, expected, use_close=False):
    """
    Compare DataFrame items by index and column and
    raise AssertionError if any item is not equal.

    Ordering is unimportant, items are compared only by label.
    NaN and infinite values are supported.

    Parameters
    ----------
    actual : pandas.DataFrame
    expected : pandas.DataFrame
    use_close : bool, optional
    If True, use numpy.testing.assert_allclose instead of
    numpy.testing.assert_equal.

    """
    if use_close:
        comp = npt.assert_allclose
    else:
        comp = npt.assert_equal

    assert (isinstance(actual, pd.DataFrame) and
            isinstance(expected, pd.DataFrame)), \
        'Inputs must both be pandas DataFrames.'

    for i, exp_row in expected.iterrows():
        assert i in actual.index, 'Expected row {!r} not 
        found.'.format(i)

        act_row = actual.loc[i]

        for j, exp_item in exp_row.iteritems():
            assert j in act_row.index, \
                'Expected column {!r} not found.'.format(j)

            act_item = act_row[j]

            if comp(act_item, exp_item):
               print("CSV files are identical") 
            else:
                print('The difference in CSV files are'.format(j, i))

actual = pd.DataFrame(df1)
expected = pd.Dataframe(df2)
assert_frames_equal(actual, expected)

我遇到以下错误：

File "/Users/saikatbhattacharjee/anaconda3/lib/python3.6/site-packages/spyder/utils/site/sitecustomize.py", line 101, in execfile
    exec(compile(f.read(), filename, 'exec'), namespace)

  File "/Users/saikatbhattacharjee/.spyder-py3/temp.py", line 52, in <module>
    actual = pd.DataFrame(df1)

  File "/Users/saikatbhattacharjee/anaconda3/lib/python3.6/site-packages/pandas/core/frame.py", line 354, in __init__
    raise ValueError('DataFrame constructor not properly called!')

ValueError: DataFrame constructor not properly called!

Answer 1

几个快速的答案（我会说这是有据可查的）

到convert 一个 dask 数据帧到一个熊猫：

actual = df1.compute()

（确保您有足够的内存来执行此操作！）

将数据写入CSV

actual.to_csv('myfiles*.csv')

每个分区写入一个文件，因为多个线程/进程不能以任何合理的方式同时写入单个文件；如果您愿意，您可以将文件（删除标题或使用过header=False）合并为一个。