Saya memiliki file csv besar, dan anggaplah itu terlihat seperti ini

ID,PostCode,Value
H1A0A1-00,H1A0A1,0
H1A0A1-01,H1A0A1,0
H1A0A1-02,H1A0A1,0
H1A0A1-03,H1A0A1,0
H1A0A1-04,H1A0A1,1
H1A0A1-05,H1A0A1,0
H1A1G7-0,H1A1G7,0
H1A1G7-1,H1A1G7,0
H1A1G7-2,H1A1G7,0
H1A1N6-00,H1A1N6,0
H1A1N6-01,H1A1N6,0
H1A1N6-02,H1A1N6,0
H1A1N6-03,H1A1N6,0
H1A1N6-04,H1A1N6,0
H1A1N6-05,H1A1N6,0
...

Saya ingin membaginya dengan nilai PostCode dan menyimpan semua baris dengan kode pos yang sama dengan CSV. saya telah mencoba

postals = data['PostCode'].unique()
for p in postals:
    df = data[data['PostCode'] == p]
    df.to_csv(directory + '/output/demographics/' + p + '.csv', header=False, index=False)

Apakah ada cara untuk melakukan ini menggunakan Dask untuk memanfaatkan multiprosesor? Terima kasih

1
Winston 3 Februari 2020, 04:06

2 jawaban

Jawaban Terbaik

Jika Anda ingin menyimpan ke parket, itu cukup mudah

Parket

import dask.dataframe as dd
import pandas as pd
import os 

fldr = 'data_pq'
data.to_parquet(fldr, partition_on="PostCode")

Ini menyimpan data untuk setiap kode pos di dalam folder bernama PostCode=xxxxxxx yang berisi file sebanyak jumlah partisi dask.dataframe Anda.

CSV

Di sini saya menyarankan Anda untuk menggunakan fungsi kustom write_file.

import dask.dataframe as dd
import pandas as pd
import os 

fldr = "data_csv"
os.makedirs(fldr, exist_ok=True)


def write_file(grp):
    pc = grp["PostCode"].unique()[0]
    grp.to_csv(f"{fldr}/{pc}.csv",
               header=False,
               index=False)
    return None


data.groupby("PostCode")\
    .apply(write_file, meta=('x', 'f8'))\
    .compute()

# the same function works for pandas df too
# data.groupby("PostCode").apply(write_file)

Anda harus memeriksa cara kerjanya berdasarkan kinerja dan akhirnya bermain dengan scheduler.

4
rpanai 4 Februari 2020, 16:23

Dask DataFrame .to_csv metode akan membuat satu file per partisi dataset Anda. Oleh karena itu, Anda dapat membiarkan Dask membuat CSV untuk Anda dengan mempartisi ulang data Anda dan menyetel kolom partisi sebagai indeks:

num_partitions = data['PostCode'].nunique().compute()
data = data.repartition(npartitions=num_partitions)
data = data.set_index('PostCode')
data.to_csv('/path/to/data/export-*.csv')

Perhatikan bahwa jika Anda memiliki kumpulan data yang sangat besar, partisi ulang dan pengaturan indeks bisa mahal dan Anda mungkin lebih baik dengan mengulangi setiap kode pos seperti yang Anda usulkan pada awalnya.

0
foglerit 3 Februari 2020, 02:23