11# Worker class
22
3- begin <- proc.time()[3 ]
4-
53# NOTE: We use "stdin" to get the process stdin instead of the command line
64inputConStdin <- file(" stdin" , open = " rb" )
75
@@ -67,11 +65,6 @@ numPartitions <- SparkR:::readInt(inputCon)
6765
6866isEmpty <- SparkR ::: readInt(inputCon )
6967
70- metadataEnd <- proc.time()[3 ]
71- dataReadEnd <- metadataEnd
72- computeEnd <- dataReadEnd
73- writeEnd <- computeEnd
74-
7568if (isEmpty != 0 ) {
7669
7770 if (numPartitions == - 1 ) {
@@ -81,15 +74,12 @@ if (isEmpty != 0) {
8174 } else {
8275 data <- readLines(inputCon )
8376 }
84- dataReadEnd <- proc.time()[3 ]
8577 output <- do.call(execFunctionName , list (splitIndex , data ))
86- computeEnd <- proc.time()[3 ]
8778 if (isOutputSerialized ) {
8879 SparkR ::: writeRawSerialize(outputCon , output )
8980 } else {
9081 SparkR ::: writeStrings(outputCon , output )
9182 }
92- writeEnd <- proc.time()[3 ]
9383 } else {
9484 if (isInputSerialized ) {
9585 # Now read as many characters as described in funcLen
@@ -98,7 +88,6 @@ if (isEmpty != 0) {
9888 data <- readLines(inputCon )
9989 }
10090
101- dataReadEnd <- proc.time()[3 ]
10291 res <- new.env()
10392
10493 # Step 1: hash the data to an environment
@@ -116,8 +105,6 @@ if (isEmpty != 0) {
116105 }
117106 invisible (lapply(data , hashTupleToEnvir ))
118107
119- computeEnd <- proc.time()[3 ]
120-
121108 # Step 2: write out all of the environment as key-value pairs.
122109 for (name in ls(res )) {
123110 SparkR ::: writeInt(outputCon , 2L )
@@ -126,7 +113,6 @@ if (isEmpty != 0) {
126113 length(res [[name ]]$ data ) <- res [[name ]]$ counter
127114 SparkR ::: writeRawSerialize(outputCon , res [[name ]]$ data )
128115 }
129- writeEnd <- proc.time()[3 ]
130116 }
131117}
132118
@@ -142,13 +128,5 @@ unlink(inFileName)
142128# Restore stdout
143129sink()
144130
145- end <- proc.time()[3 ]
146-
147- cat(" stats: total " , (end - begin ), " \n " , file = stderr())
148- cat(" stats: metadata " , (metadataEnd - begin ), " \n " , file = stderr())
149- cat(" stats: input read " , (dataReadEnd - metadataEnd ), " \n " , file = stderr())
150- cat(" stats: compute " , (computeEnd - dataReadEnd ), " \n " , file = stderr())
151- cat(" stats: output write " , (writeEnd - computeEnd ), " \n " , file = stderr())
152-
153131# Finally print the name of the output file
154132cat(outputFileName , " \n " )
0 commit comments