python - Dask 相当于 pandas.DataFrame.update

Question

我有一些函数正在使用 pandas.DataFrame.update 方法，我正在尝试使用 Dask 代替数据集，但是 Dask Pandas API 没有实现 update 方法。有没有其他方法可以在 Dask 中获得相同的结果？

以下是我使用更新的方法:

1. 前向用最后已知值填充数据

df.update(df.filter(like='/').mask(lambda x: x == 0).ffill(1))

输入

id .. .. ..(some cols) 1/1/20 1/2/20 1/3/20 1/4/20 1/5/20 1/6/20 ....
1                      10     20     0      40     0      50
2                      10     30     30     0      0      50
.
.

输出

id .. .. ..(some cols) 1/1/20 1/2/20 1/3/20 1/4/20 1/5/20 1/6/20 ....
1                      10     20     20     40     40      50
2                      10     30     30     30     30      50
.
.

1. 根据 id/index 列将数据框中的值替换为另一个数据框中的值

def replace_names(df1, df2, idxCol = 'id', srcCol = 'name', dstCol = 'name'):
    df1 = df1.set_index(idxCol)
    df1[dstCol].update(df2.set_index(idxCol)[srcCol])
    return df1.reset_index()
df_new = replace_names(df1, df2)

输入

df1

id    name  ...
123   city a
456   city b
789   city c
789   city c
456   city b
123   city a
.
.
.

df2

id    name  ...
123   City A
456   City B
789   City C
.
.
.

输出

id    name  ...
123   City A
456   City B
789   City C
789   City C
456   City B
123   City A
.
.
.

Answer 1

问题2

有一种方法可以部分解决这个问题。我假设 df2 比 df1 小得多，它实际上适合内存，因此我们可以读取为 pandas 数据帧。如果是这种情况，如果 df1 是 pandas 或 dask 数据帧，但 df2 应该是一个 pandas

import pandas as pd
import dask.dataframe as dd

def replace_names(df1, # can be pandas or dask dataframe
                  df2, # this should be pandas.
                  idxCol='id',
                  srcCol='name',
                  dstCol='name'):
    diz = df2[[idxCol, srcCol]].set_index(idxCol).to_dict()[srcCol]
    out = df1.copy()
    out[dstCol] = out[idxCol].map(diz)
    return out

问题1

关于第一个问题，以下代码适用于pandas 和dask

df = pd.DataFrame({'a': {0: 1, 1: 2},
 'b': {0: 3, 1: 4},
 '1/1/20': {0: 10, 1: 10},
 '1/2/20': {0: 20, 1: 30},
 '1/3/20': {0: 0, 1: 30},
 '1/4/20': {0: 40, 1: 0},
 '1/5/20': {0: 0, 1: 0},
 '1/6/20': {0: 50, 1: 50}})

# if you want to try with dask
# df = dd.from_pandas(df, npartitions=2)

cols = [col for col in df.columns if "/" in col]
df[cols] = df[cols].mask(lambda x: x==0).ffill(1) #.astype(int)

如果您希望输出为整数，请删除最后一行的注释。

更新问题 2 如果您想要一个仅dask 的解决方案，您可以尝试以下方法。

数据

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

df1 = pd.DataFrame({'id': {0: 123, 1: 456, 2: 789, 3: 789, 4: 456, 5: 123},
 'name': {0: 'city a',
  1: 'city b',
  2: 'city c',
  3: 'city c',
  4: 'city b',
  5: 'city a'}})

df2 = pd.DataFrame({'id': {0: 123, 1: 456, 2: 789},
 'name': {0: 'City A', 1: 'City B', 2: 'City C'}})

df1 = dd.from_pandas(df1, npartitions=2)
df2 = dd.from_pandas(df2, npartitions=2)

案例一

在这种情况下，如果一个 id 存在于 df1 但不存在于 df2 中，您将名称保留在 df1 中>.

def replace_names_dask(df1, df2,
                       idxCol='id',
                       srcCol='name',
                       dstCol='name'):
    if srcCol == dstCol:
        df2 = df2.rename(columns={srcCol:f"{srcCol}_new"})
        srcCol = f"{srcCol}_new"
    
    def map_replace(x, srcCol, dstCol):
        x[dstCol] = np.where(x[srcCol].notnull(),
                             x[srcCol],
                             x[dstCol])
        return x
    
    df = dd.merge(df1, df2, on=idxCol, how="left")
    df = df.map_partitions(lambda x: map_replace(x, srcCol, dstCol))
    df = df.drop(srcCol, axis=1)
    return df

df = replace_names_dask(df1, df2)

案例2

在这种情况下，如果一个 id 存在于 df1 但不存在于 df2 中，那么输出中的 name df 将是 NaN(与标准左连接一样)

def replace_names_dask(df1, df2,
                       idxCol='id',
                       srcCol='name',
                       dstCol='name'):
    df1 = df1.drop(dstCol, axis=1)
    df2 = df2.rename(columns={srcCol: dstCol})
    df = dd.merge(df1, df2, on=idxCol, how="left")
    return df

df = replace_names_dask(df1, df2)