left_join two dataframes out of memory

[SMousavi90]

I have recently started using this library but I have an issue while doing a join between two data.frames

a <- my first dataframe has 2127625 rows
b <- the second one has 73364 rows

after setting up the diskframe, It shows that:
The number of workers available for disk.frame is 8

a and b are converted to diskframes before doing the join:

a <- as.disk.frame(a, outdir = "/home/ruser/tmp/a.mdf", overwrite = TRUE)
b <- as.disk.frame(b, outdir = "/home/ruser/tmp/b.mdf", overwrite = TRUE) 

now at some part of my code I'm doing this join:

c <- a %>% left_join(b)

this line returns:

...
Hashing...
Hashing...
Appending disk.frames: 
Error: cannot allocate vector of size 581.5 Gb

then I tried doing it in this way:
c <- a %>% left_join(b, merge_by_chunk_id = TRUE)

firstly, it used the whole (62.8 GB) of my ram, then returned an error:

Error in unserialize(node$con) : 
  Failed to retrieve the value of MultisessionFuture (future_mapply-1) from cluster RichSOCKnode #1 (PID 25440 on localhost ‘localhost’). The reason reported was ‘error reading from connection’. Post-mortem diagnostic: A process with this PID exists, which suggests that the localhost worker is still alive.

and didn't do the join!

I also tested it with:
setup_disk.frame(workers = 16)

same result!

just to mention that I have done some joins on my other data.frames and they were done successfully but this one (which is only greater than the other data.frames) failed.

could you please help me to understand what the problem is?

[xiaodaigh]

I suspect your left join has a lot of duplicate keys. For example, if i generate a and b as below

library(disk.frame)
setup_disk.frame()

a = data.frame(
  id = sample(1:8, 2127625, replace=TRUE)
)

b = data.frame(
  id = sample(1:8, 73364, replace=TRUE)
)


library(dplyr)

left_join(a, b)

This would also fail to join on my computer and it's allocating 72.7GB of RAM.

You can't improve the situation by having more workers, but you might be able to fix the issue by having fewer workers and working on smaller chunks like this.

setup_disk.frame(workers = 4) # reduce to 1 if still not working since the more worker you have the more memory it uses
a = as.disk.frame(a)
b = as.disk.frame(b)

nchunks_for_merge = max(nchunks(a), nchunks(b))

# double the number of chunks and shard by the merge key which is "id" in my case
a = rechunk(a, nchunks=nchunks_for_merge*2, shardby = "id")
b = rechunk(b, nchunks=nchunks_for_merge*2, shardby = "id")

nc = max(nchunks(a), nchunks(b))

#setup_disk.frame(workers=1) # if necessary

system.time(ab <- left_join(a, b, merge_by_chunk_id=TRUE))