PythonのDASKを使ってpandasでは処理できない巨大CSVを前処理する方法

process-large-csv-with-dask.py

import dask.dataframe as dd

# 分割されたCSVでも一つの巨大CSVでも同じように読み込める
file1 = 'DW-exported-*.csv'
df = dd.read_csv(file1, header=0, names=('Pages','VisitNum','HitDepth','Date','CV','Customer ID','Referrer','PV'), dtype={'Pages':'object','VisitNum':'int16','HitDepth':'int16','Date':'object','CV':'object','eVar13':'object','Referrer':'object','PV':'int16'})
# カラム名や型を指定しない場合
#df = dd.read_csv(file1, header=0)

# この結果、複数のパーティションに分割されたpandas dataframeが生成される

print('生成されたパーティションの数：' + str(df.npartitions))

# フィルタや変換などの前処理はパーティション単位で並列処理される

# まずフィルタでレコード数を減らす例
df = df[df['Customer ID'].str.contains('^\d+', na=False)]

# 値を変換する例
df['Customer ID'] = df['Customer ID'].str.replace('\/.*$', '')

# NULLを含むレコードを除外する例
df = df.dropna(subset=['Pages', 'Date', 'Customer ID', 'PV'])

# カラム削除の例
df = df.drop('CV', axis=1)

# 他のCSVとマージ(JOIN)する例
file2 = 'ID一覧.csv'
loyal = dd.read_csv(file2, header=0)
merge = dd.merge(df, loyal, left_on='Customer ID', right_on='AAID')
merge = merge.drop('AAID', axis=1)

# headやtailなら全データを読み込まないので速い
merge.head()

# 処理が終わったらパーティション単位でファイル保存
merge.to_csv("DW-processed-*.csv")

# メモリに収まるほど小さくなった場合は、一つのファイルにまとめて保存できる
merge.compute().to_csv("DW-processed.csv")