Agfdhyk

I have a data frame as follows:

import org.apache.spark.sql.expressions.Window

import org.apache.spark.sql.Column

import org.apache.spark.sql.functions._

import spark.implicits._



// some data...

val df = Seq(

  (1, "AA", "BB", ("AA", "BB")),

  (2, "AA", "BB", ("AA", "BB")),

  (3, "AB", "BB", ("AB", "BB"))

).toDF("id","name", "surname", "array")



df.show()

and i am looking to calculate the edit distance between the 'array' column in successive row. As an example i want to calculate the edit distance between the 'array' entity in column 1 ("AA", "BB") and the the 'array' entity in column 2 ("AA", "BB"). Here is the edit distance function i am using:

def editDist2[A](a: Iterable[A], b: Iterable[A]): Int = {

  val startRow = (0 to b.size).toList

  a.foldLeft(startRow) { (prevRow, aElem) =>

    (prevRow.zip(prevRow.tail).zip(b)).scanLeft(prevRow.head + 1) {

      case (left, ((diag, up), bElem)) => {

        val aGapScore = up + 1

        val bGapScore = left + 1

        val matchScore = diag + (if (aElem == bElem) 0 else 1)

        List(aGapScore, bGapScore, matchScore).min

      }

    }

  }.last

}

I know i need to create a UDF for this function but can't seem to be able to. If i use the function as is and using Spark Windowing to get at the pervious row:

// creating window - ordered by ID

val window = Window.orderBy("id")



// using the window with lag function to compare to previous value in each column

df.withColumn("edit-d", editDist2(($"array"), lag("array", 1).over(window))).show()

i get the following error:

<console>:245: error: type mismatch;

 found   : org.apache.spark.sql.ColumnName

 required: Iterable[?]

       df.withColumn("edit-d", editDist2(($"array"), lag("array", 1).over(window))).show()

I figured out you can use Spark's own levenshtein function for this. This function takes in two string to compare, so it can't be used with the array.

// creating window - ordered by ID

val window = Window.orderBy("id")



// using the window with lag function to compare to previous value in each column

df.withColumn("edit-d", levenshtein(($"name"), lag("name", 1).over(window)) + levenshtein(($"surname"), lag("surname", 1).over(window))).show()

giving the desired output:

+---+----+-------+--------+------+

| id|name|surname|   array|edit-d|

+---+----+-------+--------+------+

|  1|  AA|     BB|[AA, BB]|  null|

|  2|  AA|     BB|[AA, BB]|     0|

|  3|  AB|     BB|[AB, BB]|     1|

+---+----+-------+--------+------+