title: Spark - Return Dataframe and accumulator from function
body: <p>I have a function that returns a <code>DataFrame</code> and two accumulators. Running manually with <code>spark-shell</code> (calling the function from the jar) it works as expected.  Doing a <code>.count</code> on the <code>DataFrame</code> populates the accumulator.</p>
  
  <p>But if I call the function from <code>spark-submit</code> the accumulator always remains empty.
  I have tried returning two <code>DataFrame</code>'s with the same strange behavior: Works in <code>spark-shell</code>, does not from <code>spark-submit</code>.</p>
  
  <p>Here is a possibly non functioning skeleton of my code:</p>
  
  <pre><code>import org.apache.spark.sql.{SaveMode, DataFrame, SQLContext, Row}
  import scala.collection._
  ...
  
  def process(
      sc:SparkContext,
      sqlContext:SQLContext,
      filepaths : RDD[String]
      ): ( 
  
      val logMetadataAccumulator = sc.accumulableCollection(mutable.ArrayBuffer[( String, String, Long, String, Long, Long)]())
      val logFailedRowAccumulator = sc.accumulableCollection(mutable.ArrayBuffer[( String)]())
      ...
      ...
      val logRecordsPre = logRawFlow.map(
          entry =&gt; {
              val date = """(\d\d\d\d)-(\d\d)-(\d\d)""".r
              if ( fields.length == 23 &amp;&amp; date.findFirstMatchIn(fields(2)).nonEmpty &amp;&amp; fields(22).forall(_.isDigit) &amp;&amp; fields(21).forall(_.isDigit) ) {
                  ...
                  Row( 1L, "blah" "blah", 0L )
              }
              else ( fields(0) == logMetaDataPrefix ) {
                  ...
                  logMetadataAccumulator += (fileName, logType, numLines, logSource, startTime, endTime)
                  Row( 0L, "blah" "blah", 0L )
              }
              else {
                  try { 
                      val fileName = fields(0)
                      logFailedRowAccumulator += (fileName)
                      Row( 0L, "blah" "blah", 0L )
                  }
                  catch {
                      case e: Exception =&gt; {
                          logFailedRowAccumulator += ("unknown")
                          Row( 0L, "blah" "blah", 0L )
                      }
                  }
              }
          }
      )
  
      val logRecords = logRecordsPre.filter( _.getLong(0) != 0L)
  
      val logDF = sqlContext.createDataFrame(logRecords, logSchema)
  
      ( logDF, logMetadataAccumulator, logFailedRowAccumulator )
  )
  </code></pre>