Concatenating dataframes created on different worker machines into a single dataframe

user_13106737 0
Viewed 143

I'm trying to write this very small program that is intended to run on a Dask Distributed cluster. On each worker machine there is a CSV file (just one, as of now), the function files is supposed to run on every worker and return a dataframe built from said file. However, I want to analyze them as a single dataframe. I'm trying to build a single dataframe from those two returned by the workers. How could I achieve that?

def files():
    for file in glob.glob("data*.csv"):
        df = pd.read_csv(file)
        print(df.head())
        column = df["cars"]
        max = column.max()
        print(max)
        return df


client = Client('SCHEDULER:8786')
a = client.submit(files, workers='WORKER1:8786',  pure=False)
b = client.submit(files, workers='WORKER2:8786', pure=False)
c = dd.merge(a, b, left_on=['id', 'cars'], right_on = ['id', 'cars'],  suffixes=['_1', '_2'], how="left")
#c = dd.concat([a,b])

After executing the code above two times commenting one of the C's each time keeps the process open and nothing gets returned or shown to the client.

My sample files are detailed below:

   id  cars
0   1   223
1   2   333
2   3  1933
3   4  2003
4   5   893
5   6  1233
6   7  1933
   id  cars
0  14   224
1  24   334
2  34  1934
3  44  3004
4  54   894
5  64  1234
6  74  1934

   id  cars
0   1   223
1   2   333
2   3  1933
3   4  2003
4   5   893
5   6  1233
6   7  1933

Thus my concatenated dataframe should have a len = len(df_a) + len(df_b)

dask-distributed dask python
1 Answers

Right now you are asking two workers to do the same task of iterating over files and returning the first file's dataframe. Let's make a reproducible fix for this.

This snippet is just to generate some dummy data:

import pandas as pd
import numpy as np

# let's create two dummy files
for i in range(2):
    df = pd.DataFrame(np.random.rand(10,5), columns=list("abcde"))
    df.to_csv(f'data{i}.csv', index=False)

Now, for the main task you want to submit a task for each worker to load a separate file. Then, for concatenation you can rely on dd.from_delayed which will concatenate the dataframes from the list of futures:

import dask.dataframe as dd
from dask.distributed import Client
from glob import glob

# start a client
client = Client()

def get_df(file):
    df = pd.read_csv(file)
    return df

list_files = glob('data*.csv')
# map each item in the list to get_df
futures = client.map(get_df, list_files)

# transform futures into dask dataframe (it concatenates them automatically)
ddf = dd.from_delayed(futures)

Update: sorry, I misunderstood your situation. So, you are interested in merging dataframes from different workers, so one way to do it is to modify your original code as follows:

a = client.submit(files, workers='WORKER1:8786',  pure=False)
b = client.submit(files, workers='WORKER2:8786', pure=False)
df_a = dd.from_delayed([a])
df_b = dd.from_delayed([b])
df_c = dd.merge(df_a, df_b, left_on=['id', 'cars'], right_on = ['id', 'cars'],  suffixes=['_1', '_2'], how="left")

Note that before merging you want to convert your futures into dask dataframes.

user_106935960
Related