title: SparkR dapply fails with arrays as input
body: <p>I am working with the <code>dapply</code> (and also <code>dapplyCollect</code>) functions in SparkR which fail when arrays are used as the input.</p>
  
  <p>My input data are in a SparkDataFrame read from a parquet file from a pre-existing process (I cannot change anything relating to this process). This dataframe consists of many million rows and several columns of data stored as arrays. Here is a simplified example created from an R dataframe.</p>
  
  <pre><code>&gt; R.df &lt;- data.frame(key = 1:3)
  &gt; R.df$data &lt;- list(seq(1, 10))
  &gt; R.df
    key                          data
  1   1 1, 2, 3, 4, 5, 6, 7, 8, 9, 10
  2   2 1, 2, 3, 4, 5, 6, 7, 8, 9, 10
  3   3 1, 2, 3, 4, 5, 6, 7, 8, 9, 10
  </code></pre>
  
  <p>The same dataframe converted to a SparkDataFrame:</p>
  
  <pre><code>&gt; Spark.df &lt;- createDataFrame(R.df)
  &gt; Spark.df
  SparkDataFrame[key:int, data:array&lt;int&gt;]
  </code></pre>
  
  <p>Next, we do the processing with a custom function and an error is thrown:</p>
  
  <pre><code>my.complicated.process &lt;- function(partition) {
    ## Here we would do something to each partition
    ## For this test, return data unchanged
    return(partition)
  }
  
  spark.results &lt;- dapply(Spark.df, my.complicated.process, schema(Spark.df))
  head(spark.results)
  
  Error in handleErrors(returnStatus, conn) : 
    org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 55.0 failed 1 times, most recent failure: Lost task 0.0 in stage 55.0 (TID 55, localhost, executor driver): org.apache.spark.SparkException: R computation failed with
   Error in (function (..., deparse.level = 1, make.row.names = TRUE, stringsAsFactors = default.stringsAsFactors(),  : 
    invalid list argument: all variables should have the same length
  Calls: compute -&gt; do.call -&gt; &lt;Anonymous&gt;
  Execution halted
      at org.apache.spark.api.r.RRunner.compute(RRunner.scala:113)
      at org.apache.spark.sql.execution.r.MapPartitionsRWrapper.apply(MapPartitionsRWrapper.scala:58)
      at org.apache.spark.sql.execution.r.MapPartitionsRWrapper.apply(MapPartitionsRWrapper.scala:29)
      at org.apache.spark.sql.execution.MapPartitionsExec$$anonfun$5.apply(objects.scala:188)
      at org.apache.spark.sql.execution.MapPartitionsExec$$anonfun$5.apply(objects.scala:185)
      at org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInter
  </code></pre>
  
  <p>I have managed workarounds to solve this:</p>
  
  <ol>
  <li>Following the example <a href="https://gist.github.com/clarkfitzg/08ad74e2099f51e5138b3e0ba72bdc07" rel="nofollow noreferrer">here</a>. By using <code>serialize</code> and <code>unserialize</code> to convert the array to simple binary type. This has the disadvantage of having to serialise before the SparkDataFrame is created.</li>
  <li>By converting the array to a comma-separated string e.g. by using <code>to_json</code>, then using a combination of <code>strsplit</code>, <code>paste</code> and <code>as.integer</code> in R and <code>split</code>, <code>cast</code> in SparkR to go between a numeric array and a string </li>
  <li>By using <code>explode</code> in SparkR to split the array column into multiple rows and pivoting the dataframe to get columns, then using <code>dapply</code> over these new columns.</li>
  </ol>
  
  <p>None of these solutions are optimal, so I was hoping to find a way for <code>dapply</code> to both take array types as inputs and also return array types.</p>
  
  <p>I am using R v3.6.3 with SparkR v2.4.5.</p>