在 Dask apply 中返回结构化行答案

【问题标题】：Return a structured row in Dask apply在 Dask apply 中返回结构化行
【发布时间】：2021-10-06 19:14:18
【问题描述】：

我正在对 Dask 数据框 中的所有行应用一个函数。在PySpark 中，我能够返回带有命名参数的spark.sql.Row 对象，以便为生成的DataFrame 创建结构化行。对 dask 数据帧中的行应用函数时，如何返回结构相似的行（包含列和类型）？

我正在寻找类似的东西：

# df is a dask.dataframe with a JSON blob in the `data` column

def process(row):
    json_data = json.loads(row.data)
    return Row(a=json_data["a"], b=json_data["b")

result = df.apply(
    process,
    axis=1,
).compute()

result

我看到这些行本身就是pd.Series，所以我尝试process 返回一个Series，但我得到了这个错误

AttributeError: 'Series' object has no attribute 'columns'

documentation 建议我可以在apply 中使用meta 参数：

meta: 一个空的 pd.DataFrame 或 pd.Series 匹配输出的 dtypes 和列名... [Inputs like] iterable of (name, dtype) 可以提供（注意名称的顺序应该匹配列的顺序）

但是，当我按照建议使用元元组的 iterable 时

result = df.apply(
    process,
    axis=1,
    meta=[("a", "int")]
).compute()

它需要一个 DataFrame 对象并返回此错误

AttributeError: 'DataFrame' object has no attribute 'name'

【问题讨论】：

您可能会发现这个答案很有帮助：stackoverflow.com/a/31364127/11390523

标签： python pandas dataframe dask dask-dataframe

【解决方案1】：

这是一个围绕 Pandas 函数开发的 dask 包装器 here：

# see unutbu's answer here: https://stackoverflow.com/a/25512372/10693596
import json
def json_to_series(text):
    keys, values = zip(*[item for dct in json.loads(text) for item in dct.items()])
    return pd.Series(values, index=keys)


def process_chunk(df):
    _tmp = df['data'].apply(json_to_series)
    return pd.concat([df, _tmp], axis=1)

result = df.map_partitions(process_chunk).compute()

【讨论】：