使用 pyarrow 在工作人员之间共享对象答案

【问题标题】：Sharing objects across workers using pyarrow使用 pyarrow 在工作人员之间共享对象
【发布时间】：2019-07-02 02:03:00
【问题描述】：

我想将共享 DataFrame 的只读访问权限授予multiprocessing.Pool.map() 创建的多个工作进程。

我想避免复制和酸洗。

我知道 pyarrow 可以用于此。但是，我发现他们的文档非常繁琐。谁能提供一个例子来说明如何做到这一点？

【问题讨论】：

pyarrow 看起来并不能让您跨进程共享数据帧。它提供了一个类似于的接口，可以跨进程共享的数据帧。如果您想避免复制/酸洗，则需要使用multiprocessing.sharedctypes，甚至还有一个创建共享结构数组以共享结构化数据的示例。
从我从这次谈话中了解到：youtu.be/Hqi_Bw_0y8Q，pyarrow 应该提供一个 pandas 集成，允许共享 DataFrame，而无需复制或序列化，不仅跨进程，甚至跨语言、框架和操作系统。
看来我们都至少部分正确。在您分享的视频的 21.20，Wes 谈到了在多个 python 进程之间共享数据的实际过程。似乎您需要设置一个 apache spark 实例来根据需要实际保存数据和数据中的 pyarrow 流（读取：序列化和复制）。这与 python 多处理共享对象非常相似，但序列化格式的开销要低得多，性能要高得多。

标签： python pandas python-multiprocessing pyarrow

【解决方案1】：

https://github.com/apache/arrow/blob/master/python/examples/plasma/sorting/sort_df.py 的示例是一个工作示例，它使用 Python 多处理在多个工作人员之间共享 Pandas 数据帧（请注意，它需要您构建一个小型 Cython 库才能运行它）。

数据帧通过Arrow's Plasma object store共享。

如果您不依赖 Python 多处理，则可以使用 Ray 以更简单的语法做您想做的事情。

要授予多个工作人员对 Pandas 数据框的只读访问权限，您可以执行以下操作。

import numpy as np
import pandas
import ray

ray.init()

df = pandas.DataFrame(np.random.normal(size=(1000, 10)))

@ray.remote
def f(df):
    # This task will run on a worker and have read only access to the 
    # dataframe. For example, "df.iloc[0][0] = 1" will raise an exception.
    try:
        df.iloc[0][0] = 1
    except ValueError:
        pass
    return df.iloc[0][0]

# Serialize the dataframe with pyarrow and store it in shared memory.
df_id = ray.put(df)

# Run four tasks that have access to the dataframe.
result_ids = [f.remote(df_id) for _ in range(4)]

# Get the results.
results = ray.get(result_ids)

注意df_id = ray.put(df)这行可以省略（也可以直接调用f.remote(df)）。在这种情况下，df 仍将存储在共享内存中并与工作人员共享，但会存储 4 次（每次调用 f.remote(df) 一次），效率较低。

【讨论】：

哇。这实际上是 Plasma 示例中相当多的样板代码 :( 我想知道 put_df() 是否正在创建副本或对数据执行一些转换，因为它在 cmets 中说“序列化”。
你说得对，put_df 和 get_dfs 现在实际上可以更简洁了。例如，现在可以使用client.put(df) 代替put_df(df)，而可以使用client.get(object_ids) 代替get_df(object_ids)。
我现在才检查了 Ray，它看起来是多处理和“原始”Arrow 的绝佳替代品。事实上，它在下面使用箭头，我不知道。所以，看起来这个答案正是我想要的。
我假设ray.put(df) 将DataFrame 中的数据复制到共享内存。是否可以以完全避免这种复制的方式创建 DataFrame？如果我理解正确，那就是 Arrow 的承诺 :)
ray.put(df) 将数据复制到共享内存是对的。假设 dataframe 最初是由 pandas 创建的，我们至少需要一个副本来将其转换为 Arrow 格式（在 Ray 中，这个序列化步骤与复制到共享内存中相结合）。另一方面，如果能够以箭头格式创建数据框（无需通过 Pandas），那么应该可以摆脱这个副本。