Как сопоставить строку с Seq <String>в Spark на Java

Question

Я хочу использовать свой собственный токенизатор для токенизации текста, хранящегося как Dataset<String>, и получить Dataset<Seq<String>> (чтобы я мог передать его CountVectorizer).

Ожидаемый ввод (/tmp/fulltext.txt):

t1 t2 t3
t4 t5

Ожидаемый результат:

[t1, t2, t3]
[t4, t5]

Я пишу токенизатор (в основном то, что он делает сейчас, то же самое, что Tokenizer, поставляемый с Spark, но мне нужнопереписать его для поддержки токенизации текста на китайском языке, поэтому я не могу использовать официальный Tokenizer):

public class Utils {

  public static Seq<String> segment(String text) {
    String[] array = text.split(" ");
    List<String> tokens = new ArrayList<>();
    for (String term : array) {
      tokens.add(term.toLowerCase());
    }
    return JavaConverters
        .asScalaIteratorConverter(tokens.iterator())
        .asScala()
        .toSeq();
  }

}

Приложение Spark, которое я пытаюсь сделать, -

public class TokenizeTest {

  public static void main(String[] args) {

    SparkSession spark = SparkSession
        .builder()
        .appName("Tokenize Test")
        .getOrCreate();

    Dataset<String> rawText = spark
        .read()
        .textFile("/tmp/fulltext.txt")
        .cache();

    Encoder<Seq> listEncoder = Encoders.bean(Seq.class);

    // Compilation error
    Dataset<Seq<String>> newText = rawText
        .map((MapFunction<String, Seq<String>>) s -> Utils.segment(s), listEncoder);

    newText.show();
    spark.stop();
  }
}

I 'Я новичок в Spark, и приведенный выше код как раз и будет работать (после прочтения официального руководства).Но оказывается, что код TokenizeTest вообще не компилируется.Как вы думаете, есть способ это исправить?

Answer 1

Использование таких коллекций Scala не сработает.Один раз Seq не является совместимым с Bean-компонентом, второй - общим.

Если вы хотите разделить, просто используйте массивы с segement, определенным как:

public class Utils {

  public static String[] segment(String text) {
    return text.split(" ");
  }

}

и TokenizeTestкак:

public class TokenizeTest {

  public static void main(String[] args) {

    SparkSession spark = SparkSession
        .builder()
        .appName("Tokenize Test")
        .getOrCreate();

    Dataset<String> rawText = spark
        .read()
        .textFile("/path/to/file")
        .cache();

    Encoder<String []> listEncoder = spark.implicits().newStringArrayEncoder();


    Dataset<String []> newText = rawText
        .map((MapFunction<String, String []>) s -> Utils.segment(s), listEncoder);

    newText.show();
    spark.stop();
  }
}

На практике вы можете вместо * изобретать колесо org.apache.spark.sql.functions.split или org.apache.spark.ml.feature.Tokenizer.