python - 读取进程并与 dask 并行连接 pandas 数据帧

Question

我正在尝试并行读取和处理 csv 文件列表，并将输出连接到单个 pandas dataframe 中以进行进一步处理。

我的工作流程包括 3 个步骤:

• 通过读取 csv 文件列表(所有文件都具有相同的结构)创建一系列 pandas 数据框

  def loadcsv(文件名):
df = pd.read_csv(文件名)
返回 df


• 通过处理 2 个现有列为每个数据框创建一个新列

  def makegeom(a,b):
返回“点(%s %s)”%(a，b)


  def applygeom(df):
df['Geom']= df.apply(lambda 行:makegeom(row['Easting'],
行 ['北']),
轴=1)
返回 df


• 将所有数据帧连接到一个数据帧中

  帧 = []
对于我在 csvtest 中:
df = applygeom(loadcsv(i))
frames.append(df)
mergedresult1 = pd.concat(帧)


在我的工作流程中，我使用了 pandas(每个 csv (15) 文件有超过 >> 2*10^6 个数据点)所以需要一段时间才能完成。我认为这种工作流程应该利用一些并行处理(至少对于 read_csv 和 apply 步骤)所以我尝试了 dask，但我做不到正确使用它。在我的尝试中，我的速度没有任何提高。

我做了一个简单的笔记本来复制我正在做的事情:

https://gist.github.com/epifanio/72a48ca970a4291b293851ad29eadb50

我的问题是......使用 dask 完成我的用例的正确方法是什么？

Answer 1

Pandas

在 Pandas 中我会使用 apply 方法

In [1]: import pandas as pd

In [2]: df = pd.DataFrame({'a': [1, 2, 3], 'b': [3, 2, 1]})

In [3]: def makegeom(row):
   ...:      a, b = row
   ...:      return 'Point(%s %s)' % (a, b)
   ...: 

In [4]: df.apply(makegeom, axis=1)
Out[4]: 
0    Point(1 3)
1    Point(2 2)
2    Point(3 1)
dtype: object

Dask.dataframe

在 dask.dataframe 中你可以做同样的事情

In [5]: import dask.dataframe as dd

In [6]: ddf = dd.from_pandas(df, npartitions=2)

In [7]: ddf.apply(makegeom, axis=1).compute()
Out[7]: 
0    Point(1 3)
1    Point(2 2)
2    Point(3 1)

添加新系列

无论哪种情况，您都可以将新系列添加到数据框中

df['geom'] = df[['a', 'b']].apply(makegeom)

创建

如果你有 CSV 数据，那么我会使用 dask.dataframe.read_csv 函数

ddf = dd.read_csv('filenames.*.csv')

如果您有其他类型的数据，那么我会使用 dask.delayed