apache-spark - Apache Spark での Jaro-Winkler スコアの計算

Question

Apache Spark Datasetの文字列全体で Jaro-Winkler 距離計算を実装する必要があります。Spark は初めてで、Web で検索してもあまり見つかりません。ご案内いただければ幸いです。flatMapを使用することを考えましたが、役に立たないことに気付き、いくつかの foreach ループを使用しようとしましたが、先に進む方法を理解できませんでした。各文字列をすべてと比較する必要があるためです。以下のデータセットのように。

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"));

上記のデータフレームで見つかったすべての文字列の jaro winkler スコアの例。

ラベル間の距離スコア、0,1 -> 0.56
ラベル間の距離スコア、0,2 -> 0.77
ラベル間の距離スコア、0,3 -> 0.45
ラベル間の距離スコア、1,2 -> 0.77
ラベル間の距離スコア、2 ,3 -> 0.79

    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<Row> 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<Row> sentenceDataFrame = spark.createDataFrame(data, schema);

            sentenceDataFrame.foreach();

        }

    }