title: No TypeTag available
body: <p>I've tried to make a generic function <code>diff</code> (like a function diff in R language) which make a difference vector of given DataFrame's target column according to given binary function, like this:</p>
  
  <ul>
  <li><p>given data: <code>df(col1, col2)</code>  </p>
  
  <p>Seq(<br>
   ("A", 1),<br>
   ("A", 1),<br>
   ("A", 5),<br>
   ("B", 1),<br>
   ("B", 3)).toDF</p></li>
  <li><p>given binary function: <code>f</code>  </p>
  
  <p>(x: Int, y: Int) => y - x</p></li>
  <li><p><code>diff(df, "col1", "col2", f, 0)</code></p></li>
  <li><p>result  </p>
  
  <p>Seq(<br>
   ("A", 1, 0),<br>
   ("A", 1, 0),<br>
   ("A", 5, 4),<br>
   ("B", 1, 0),<br>
   ("B", 3, 2)).toDF  </p></li>
  </ul>
  
  <p>key difference with <code>diff</code> in R is "it works <code>group by</code> manner"    </p>
  
  <p>anyway, on compile, there occurs an error like this   </p>
  
  <p><code>Error:(41, 22) No TypeTag available for Array[B]
      val funcUdf = udf(func)</code>  </p>
  
  <p>and something like <code>udf[Array[B], Array[A]](seqFuncApply)</code> is not a solution...</p>
  
  <pre><code>  import org.apache.spark.sql._
    import org.apache.spark.sql.functions._
    import scala.reflect.ClassTag
  
    val spark = SparkSession.builder().appName("sparksql").master("local").getOrCreate()
    import spark.implicits._
  
    def diff[A: ClassTag, B: ClassTag](df: DataFrame, key: String, target: String, diffFunc: (A, A) =&gt; B, zero: B) = {
  
      val seqFuncApply =
        (xs: Array[A]) =&gt; {
          if (xs.length &lt; 2) Array(zero)
          else xs.tail.zipWithIndex.map { tu =&gt;
            val x2 = tu._1
            val idx: Int = tu._2
            val x1 = xs.init(idx)
            diffFunc(x2, x1)
          }.+:(zero)
        }
  
      val funcUdf = udf(seqFuncApply)
  
      val resultDf: DataFrame =
        df.select(key, target)
          .rdd
          .map(row =&gt; (row.getAs[A](0), row.getAs[A](1)))
          .aggregateByKey(Array[A]())(_ :+ _, _ ++ _)
          .toDF(key, target)
          .withColumn("diff_" + target, funcUdf(col(target)))
  
      val cbind: (DataFrame, DataFrame) =&gt; DataFrame =
        (df, df2) =&gt; {
          val x =
            df.withColumn("primaryKeyForCbind", monotonically_increasing_id())
              .withColumn("orderKeyForCbind", monotonically_increasing_id()).as("df")
          val y =
            df2.withColumn("primaryKeyForCbind", monotonically_increasing_id()).as("df2")
          x.join(y, col("df.primaryKeyForCbind") === col("df2.primaryKeyForCbind"))
            .sort("orderKeyForCbind")
            .drop("primaryKeyForCbind", "orderKeyForCbind")
        }
  
      cbind(
        resultDf.select(col(key), explode(col(target))).as("target"),
        resultDf.select(explode(col("diff_" + target)).as("diff_" + target)))
    }
  </code></pre>