3
\$\begingroup\$

I have a dataset with 16 million rows and may increase upwards of 30 million. I am using the parLapply to run across three cores in R. But it's taking two days to run to completion. When I try smaller datasets of about 60,000 it takes less than 5 minutes to run, what may be cause of this disparity.

Desktop Specs : Corei5 -QuadCore , 4GB RAM

FG DataSet (16 million rows)

"Id","R","T"
"1","12","43963"
"2","12","50273"
"3","12","40805" 
"4","13","50273"
"5","13","40805"
"6","14","40805"

AB (1.3 million rows)

"Id","R",
"1","12"   
"2","13"
"3","14"   
"4","15"

Locations (6600 rows)

T,NEWLong,NEWLat,SITENAME,
43963,-77.108995,17.942062,HARBOUR TOWN
50273,-77.108995,17.942062,NEW MEADOWS
40805,-77.108995,17.942062,ISLE AVENUE

Code

num_cores = detectCores() -1
cl = makeCluster(num_cores)
clusterExport(cl,varlist = c("FG","AB","sites","distancematrix")
              ,envir=environment())

results = parLapply(cl,1:nrow(AB),function(i){
  row = AB[i,2]
  filtered = subset(FG,FG$R == AB[i,2])
  sites = merge(filtered , locations , by.x = "T" , by.y = "T" , all.x = FALSE)
  resultdf =unique(data.frame(sites$NAME,sites$NEWLong,sites$NEWLat))

    if ((nrow(resultdf))==0)
   {

    VAL = data.frame("AN" = AB[i,2] ,"SCORE" = 0 ,"SITES" = 0,"DISTANCE" = 0)
   }
  else if ((nrow(resultdf) > 0) & (nrow(resultdf) < 4))
  {

  alldistance = round(distanceMatrix(resultdf))
  VAL2 = data.frame("AN" = AB[i,2] ,"SCORE"= 1 ,"SITES" =       nrow(resultdf),"DISTANCE"=sum(alldistance))
   }
   else if ((nrow(resultdf) >= 4) & (nrow(resultdf) <= 10 ))
  {

  alldistance = round(distanceMatrix(resultdf))
  if (sum(alldistance) == 0)
  {
    VAL = data.frame("AN" = AB[i,2] ,"SCORE"= 1 ,"SITES" = nrow(resultdf),"DISTANCE"=sum(alldistance))
  }
  else
  {
    value = nrow(resultdf)-1
    require(fpc)
    clustervaluePAMK = pamk(alldistance,krange = 1:value, criterion = "asw" ,critout = TRUE , usepam=FALSE, ns = 2)
    clustervaluePAMK  = clustervaluePAMK$nc
    VAL2 = data.frame("AN" = AB[i,2] ,"SCORE"= clustervaluePAMK ,"SITES" = nrow(resultdf),"DISTANCE"=sum(alldistance))

  }

}
else 
{
  alldistance = round(distanceMatrix(resultdf))
  if (sum(alldistance) == 0)
  {
    VAL = data.frame("AN" = AB[i,2] ,"SCORE"= 1 ,"SITES" =     nrow(resultdf),"DISTANCE"=sum(alldistance))
  }
  else
  {
    require(fpc)
    clustervaluePAMK = pamk(alldistance,krange = 1:10, criterion = "asw" ,critout = TRUE , usepam=FALSE, ns = 2)
    clustervaluePAMK  = clustervaluePAMK$nc
    VAL = data.frame("AN" = AB[i,2] ,"SCORE"= clustervaluePAMK ,"SITES" = nrow(resultdf),"DISTANCE"=sum(alldistance))

  }

}

})

{FGL <- merge(FG, locations) }) and object.size(FGL)

 user  system elapsed 
 393.70   10.24  993.51 

656225664 bytes

Code Profile --For this section I ran against 60,000 elements.

$by.self
               self.time self.pct total.time total.pct
"unserialize"     174.70    99.99     174.70     99.99
"as.character"      0.02     0.01       0.02      0.01

$by.total
                     total.time total.pct self.time self.pct
"clusterApply"           174.72    100.00      0.00     0.00
"do.call"                174.72    100.00      0.00     0.00
"lapply"                 174.72    100.00      0.00     0.00
"parLapply"              174.72    100.00      0.00     0.00
"staticClusterApply"     174.72    100.00      0.00     0.00
"unserialize"            174.70     99.99    174.70    99.99
"FUN"                    174.70     99.99      0.00     0.00
"recvData"               174.70     99.99      0.00     0.00
"recvData.SOCKnode"      174.70     99.99      0.00     0.00
"as.character"             0.02      0.01      0.02     0.01
"cut"                      0.02      0.01      0.00     0.00
"cut.default"              0.02      0.01      0.00     0.00
"factor"                   0.02      0.01      0.00     0.00
"split"                    0.02      0.01      0.00     0.00
"split.default"            0.02      0.01      0.00     0.00
"splitIndices"             0.02      0.01      0.00     0.00
"splitList"                0.02      0.01      0.00     0.00
"structure"                0.02      0.01      0.00     0.00

$sample.interval
[1] 0.02

$sampling.time
[1] 174.72
\$\endgroup\$
  • \$\begingroup\$ @flodel added runtime \$\endgroup\$ – user2757228 Aug 3 '16 at 12:06
  • \$\begingroup\$ @flodel I might have to rethink that sections as to see if there is a better way to process the data. Might have to do it in chunks or make use of some of the bigmemory library and see what happens \$\endgroup\$ – user2757228 Aug 4 '16 at 11:55
  • 1
    \$\begingroup\$ For large data, data.table is sooo much faster than regular data frames. \$\endgroup\$ – ff524 Aug 5 '16 at 9:07
  • \$\begingroup\$ @flodel I am already doing a premerge since working with the data tables are much quicker and using setkeys to allow for faster search. All this happens out side the loop. \$\endgroup\$ – user2757228 Aug 10 '16 at 12:01
  • \$\begingroup\$ I have rolled back the question to Rev 4. Please see What to do when someone answers. \$\endgroup\$ – 200_success Aug 11 '16 at 22:14
4
\$\begingroup\$

Hard to tell without sample data, but lets start with a cleaned up version as there's soo much duplicated code here and the formatting is inconsistent.

  • The require should probably go to the top?
  • AB[i, 2], nrow(resultdf) are run more than once and that's not good.
  • Some expressions are the same in multiple branches and can be merged, e.g. alldistance = ... and sum(alldistance).
  • AFAIK parLapply just uses the return value of the function, so the assignments to VAR and VAR2 are super confusing.

All refactored that looks like this now, which almost fits into a single screen now:

require(fpc)

cl = makeCluster(detectCores() - 1)
clusterExport(cl, varlist = c("FG", "AB", "sites", "distancematrix"), envir = environment())

results = parLapply(cl, 1:nrow(AB), function(i) {
  row = AB[i, 2]
  filtered = subset(FG, FG$R == row)
  sites = merge(filtered, locations, by.x = "T", by.y = "T", all.x = FALSE)
  resultdf = unique(data.frame(sites$NAME, sites$NEWLong, sites$NEWLat))

  n = nrow(resultdf)

  if (n == 0)
  {
    data.frame("AN" = row, "SCORE" = 0, "SITES" = 0, "DISTANCE" = 0)
  }
  else
  {
    alldistance = round(distanceMatrix(resultdf))
    s = sum(alldistance)

    if (n < 4 || s == 0)
    {
      score = 1
    }
    else
    {
      score = pamk(alldistance, krange = 1:min(n - 1, 10), criterion = "asw", critout = TRUE, usepam = FALSE, ns = 2)$nc
    }

    data.frame("AN" = row, "SCORE" = score, "SITES" = n, "DISTANCE" = s)
  }
})
\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.