title: Jaro-Winkler score calculation in Apache Spark
body: <p>We need to implement Jaro-Winkler distance calculation across string in Apache Spark <strong>Dataset</strong>. We are new to spark and after searching in web we are not able to find much. It would be great if you can guide us. We thought of using <strong>flatMap</strong> then realized it won’t help, then we tried to use couple of foreach loops but not able to figure how to go forward. As each of the string has to be compared against all. Like in the below dataset.</p>
  
  <pre><code>RowFactory.create(0, "Hi I heard about Spark"),
  RowFactory.create(1,"I wish Java could use case classes"),
  RowFactory.create(2,"Logistic,regression,models,are,neat"));
  </code></pre>
  
  <p>Example jaro winkler score across all string found in the above dataframe.<br></p>
  
  <blockquote>
    <p>Distance score between label, 0,1 -> 0.56<br> Distance score between
    label, 0,2 -> 0.77<br> Distance score between label, 0,3 -> 0.45<br>
    Distance score between label, 1,2 -> 0.77<br> Distance score between
    label, 2,3 -> 0.79<br></p>
  </blockquote>
  
  <pre><code>    import java.util.Arrays;
      import java.util.Iterator;
      import java.util.List;
  
      import org.apache.spark.SparkConf;
      import org.apache.spark.api.java.JavaSparkContext;
      import org.apache.spark.api.java.function.FlatMapFunction;
      import org.apache.spark.sql.Dataset;
      import org.apache.spark.sql.Row;
      import org.apache.spark.sql.RowFactory;
      import org.apache.spark.sql.SQLContext;
      import org.apache.spark.sql.SparkSession;
      import org.apache.spark.sql.types.DataTypes;
      import org.apache.spark.sql.types.Metadata;
      import org.apache.spark.sql.types.StructField;
      import org.apache.spark.sql.types.StructType;
  
      import info.debatty.java.stringsimilarity.JaroWinkler;
  
      public class JaroTestExample {
       public static void main( String[] args )
          {
        System.setProperty("hadoop.home.dir", "C:\\winutil");
        JavaSparkContext sc = new JavaSparkContext(new SparkConf().setAppName("SparkJdbcDs").setMaster("local[*]"));
        SQLContext sqlContext = new SQLContext(sc);
        SparkSession spark = SparkSession.builder()
          .appName("JavaTokenizerExample").getOrCreate();
         JaroWinkler jw = new JaroWinkler();
  
              // substitution of s and t
              System.out.println(jw.similarity("My string", "My tsring"));
  
              // substitution of s and n
              System.out.println(jw.similarity("My string", "My ntrisg"));
  
              List&lt;Row&gt; data = Arrays.asList(
          RowFactory.create(0, "Hi I heard about Spark"),
          RowFactory.create(1,"I wish Java could use case classes"),
          RowFactory.create(2,"Logistic,regression,models,are,neat"));
  
              StructType schema = new StructType(new StructField[] {
        new StructField("label", DataTypes.IntegerType, false,
          Metadata.empty()),
        new StructField("sentence", DataTypes.StringType, false,
          Metadata.empty()) });
  
              Dataset&lt;Row&gt; sentenceDataFrame = spark.createDataFrame(data, schema);
  
              sentenceDataFrame.foreach();
  
          }
  
      }
  </code></pre>