时间戳

首页 > 解决方案 > 在镶木地板存储中添加到 dask 数据帧

pandas - 在镶木地板存储中添加到 dask 数据帧

问题描述

将数据（pandas 数据帧）添加到镶木地板存储中现有的 dask 数据帧的推荐方法是什么？

例如，此测试会间歇性地失败：

import dask.dataframe as dd
import numpy as np
import pandas as pd

def test_dask_intermittent_error(tmp_path):
    df = pd.DataFrame(np.random.randn(100, 1), columns=['A'],
                      index=pd.date_range('20130101', periods=100, freq='T'))
    dfs = np.array_split(df, 2)

    dd1 = dd.from_pandas(dfs[0], npartitions=1)
    dd2 = dd.from_pandas(dfs[1], npartitions=1)

    dd2.to_parquet(tmp_path)

    _ = (dd1
         .append(dd.read_parquet(tmp_path))
         .to_parquet(tmp_path))

    assert_frame_equal(df,
                       dd.read_parquet(tmp_path).compute())

给

.venv/lib/python3.7/site-packages/dask/dataframe/core.py:3812: in to_parquet
    return to_parquet(self, path, *args, **kwargs)
...
fastparquet.util.ParquetException: Metadata parse failed: /private/var/folders/_1/m2pd_c9d3ggckp1c1p0z3v8r0000gn/T/pytest-of-jfaleiro/pytest-138/test_dask_intermittent_error0/part.0.parquet

我们考虑依赖一个简单的附加并在检索后找出顺序，但这似乎遇到了一个不同的错误，即：

def test_dask_prepend_as_append(tmp_path):
    df = pd.DataFrame(np.random.randn(100, 1), columns=['A'],
                      index=pd.date_range('20130101', periods=100, freq='T'))
    dfs = np.array_split(df, 2)

    dd1 = dd.from_pandas(dfs[0], npartitions=1)
    dd2 = dd.from_pandas(dfs[1], npartitions=1)

    dd2.to_parquet(tmp_path)
    dd1.to_parquet(tmp_path, append=True)

    assert_frame_equal(df,
                       dd.read_parquet(tmp_path).compute())

给

ValueError: Appended divisions overlapping with previous ones.

标签： pandasdaskdask-distributed

解决方案

如果您在写入时避免使用“_metadata”文件（您将使用默认设置和 pyarrow），那么您可以简单地重命名文件，以确保当 glob 列出时，前置分区出现在其余分区之前。通常，Dask 会以序列号 0 开始命名。

推荐阅读