Java Исключение в Spark UDF.

Question

Проблемы с литьем при работе с искрой UDF

import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.gson.Gson;
import com.typesafe.config.Config;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.sql.*;
import org.apache.spark.sql.api.java.UDF1;
import org.apache.spark.sql.types.DataTypes;
import scala.Tuple2;
import scala.collection.JavaConverters;
import scala.collection.mutable.WrappedArray;


import java.io.IOException;
import java.io.Serializable;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ExecutionException;
import java.util.stream.Collectors;
import static java.util.Arrays.copyOfRange;
import static org.apache.spark.sql.functions.*;


  UDF1 mode1 = new UDF1<WrappedArray<Map<Double, Integer>>, String>() {
  @Override
  public String call(WrappedArray<Map<Double, Integer>> maps) throws Exception {

    List<Map<Double, Integer>> lis = (List<Map<Double, Integer>>) JavaConverters.seqAsJavaListConverter(maps).asJava();
    java.util.Map<Double,Integer> a= lis.stream().flatMap(map -> map.entrySet().stream())
            .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
    System.out.println(a.get(key));
    return "";
  }
};

Ошибка:

Причина: java .lang.ClassCastException: scala .collection.immutable .Map $ Map1 нельзя преобразовать в java .util.Map в java .util.stream.ReferencePipeline $ 7 $ 1.accept (ReferencePipeline. java: 269)