【发布时间】：2019-09-30 18:51:48
【问题描述】：
我是 Dask 的新手，正在寻找一种方法来展平 PANDAS 数据框中的字典列。这是 1600 万行数据帧的第一行的屏幕截图：
这里是三行文本的示例：
{{u'F9_07_PZ_COMP_DIRECT': u'0', u'F9_07_PZ_DIRTRSTKEY_NAME': u'DEBRA MEALY', u'F9_07_PZ_COMP_OTHER': u'0', u'F9_07_PZ_COMP_RELATED': u'0', u'F9_07_PZ_TITLE': u'CHAIR PERSON', u'F9_07_PZ_AVE_HOURS_WEEK': u'1.00', u'F9_07_PC_TRUSTEE_INDIVIDUAL': u'X'}, {u'F9_07_PZ_COMP_DIRECT': u'0', u'F9_07_PZ_DIRTRSTKEY_NAME': u'HELEN GORDON', u'F9_07_PZ_COMP_OTHER': u'0', u'F9_07_PZ_COMP_RELATED': u'0', u'F9_07_PZ_TITLE': u'VICE CHAIR', u'F9_07_PZ_AVE_HOURS_WEEK': u'1.00', u'F9_07_PC_TRUSTEE_INDIVIDUAL': u'X'}, {'F9_07_PC_HIGH_COMP_EMPLOYEE': 'X', 'F9_07_PZ_DIRTRSTKEY_NAME': 'ROB S KHANUJA', 'F9_07_PZ_COMP_OTHER': '14902', 'F9_07_PZ_COMP_RELATED': '0', 'F9_07_PZ_TITLE': 'EXEC. DIR. OPERATIONS', 'F9_07_PZ_AVE_HOURS_WEEK': '40.00', 'F9_07_PZ_COMP_DIRECT': '133173'}}
我通常会使用以下代码展平 Form990PartVIISectionAGrp 列：
df = pd.concat([df.drop(['Form990PartVIISectionAGrp'], axis=1), df['Form990PartVIISectionAGrp'].swifter.apply(pd.Series)], axis=1)
我希望在 Dask 中执行此操作，但收到以下错误：“ValueError：计算数据中的列与提供的元数据中的列不匹配。”
我正在使用 Python 2.7。我导入相关包
from dask import dataframe as dd
from dask.multiprocessing import get
from multiprocessing import cpu_count
nCores = cpu_count()
为了测试代码，我创建了一个随机数据样本：
dfs = df.sample(1000)
然后生成Dask数据框：
ddf = dd.from_pandas(dfs, npartitions=nCores)
该列当前为字符串格式，因此我将其转换为字典。通常，我只会写一行代码：
dfs['Form990PartVIISectionAGrp'] = dfs['Form990PartVIISectionAGrp'].apply(literal_eval)
但我改为尝试以更“类似 Dask”的形式在此处执行此操作，因此我编写了以下函数然后应用它：
def make_dict(dfs):
dfs['Form990PartVIISectionAGrp'] = dfs['Form990PartVIISectionAGrp'].apply(literal_eval)
return dfs
ddf_out = ddf.map_partitions(make_dict, meta=dfs[:0]).compute()
这行得通——它返回一个 PANDAS 数据框，其中 Form990PartVIISectionAGrp 列是字典格式（但是它并不比非 Dask 应用快）。
然后我重新创建 Dask DF：
ddf = dd.from_pandas(ddf_out, npartitions=nCores)
并编写一个函数来展平列：
def flatten(ddf_out):
ddf_out = pd.concat([ddf_out.drop(['Form990PartVIISectionAGrp'], axis=1), ddf_out['Form990PartVIISectionAGrp'].apply(pd.Series)], axis=1)
#ddf_out = ddf_out['Form990PartVIISectionAGrp'].apply(pd.Series)
return ddf_out
如果我再运行这段代码：
result = ddf.map_partitions(flatten)
我得到以下输出，其中列没有被展平：
我还收到有关缺少元数据的错误，并且鉴于上述内容无助于解析字典列，因此我创建了一个由普通 Python 展平列生成的列列表，并使用它来创建字典列和数据类型：
metadir = {u'BusinessName': 'O', u'F9_07_PC_FORMER': 'O', u'F9_07_PC_HIGH_COMP_EMPLOYEE': 'O',
u'F9_07_PC_KEY_EMPLOYEE': 'O', u'F9_07_PC_OFFICER': 'O',
u'F9_07_PC_TRUSTEE_INDIVIDUAL': 'O', u'F9_07_PC_TRUSTEE_INSTITUTIONAL': 'O',
u'F9_07_PZ_AVE_HOURS_WEEK': 'O', u'F9_07_PZ_AVE_HOURS_WEEK_RELATED': 'O',
u'F9_07_PZ_COMP_DIRECT': 'O', u'F9_07_PZ_COMP_OTHER': 'O',
u'F9_07_PZ_COMP_RELATED': 'O', u'F9_07_PZ_DIRTRSTKEY_NAME': 'O',
u'F9_07_PZ_TITLE': 'O', u'NameBusiness': 'O', u'URL': 'O'}
然后我用这个元数据应用 flatten 函数：
result = ddf.map_partitions(flatten, meta=metadir)
我得到以下输出结果：
运行 result.columns 会产生这样的结果：
失败的地方是运行 compute()，我收到以下错误消息：“ValueError：计算数据中的列与提供的元数据中的列不匹配。”无论我写，我都会得到同样的错误：
result.compute()
或
result.compute(meta=metadir)
我不确定我在这里做错了什么。 result 中的列似乎与 metadir 中的列相匹配。任何建议将不胜感激。
更新： 这是我更新 flatten 功能的尝试。
meta = pd.DataFrame(columns=['URL', 'F9_07_PC_TRUSTEE_INDIVIDUAL',
'F9_07_PZ_DIRTRSTKEY_NAME',
'F9_07_PZ_COMP_OTHER',
'F9_07_PZ_COMP_RELATED',
'F9_07_PZ_TITLE',
'F9_07_PZ_AVE_HOURS_WEEK',
'F9_07_PZ_COMP_DIRECT',
'F9_07_PZ_AVE_HOURS_WEEK_RELATED',
'F9_07_PC_OFFICER',
'F9_07_PC_HIGH_COMP_EMPLOYEE',
'BusinessName',
'F9_07_PC_KEY_EMPLOYEE',
'F9_07_PC_TRUSTEE_INSTITUTIONAL',
'NameBusiness',
'F9_07_PC_FORMER'], dtype="O")
def flatten(ddf_out):
ddf_out = pd.concat([df.drop(['Form990PartVIISectionAGrp'], axis=1), df['Form990PartVIISectionAGrp'].apply(pd.Series)], axis=1)
for m in meta:
if m not in ddf_out:
df[m] = ''
return ddf_out
然后我运行：
result = ddf.map_partitions(flatten, meta=meta).compute()
【问题讨论】：
-
以上是一些示例数据。
标签： python pandas dask flatten