Исключение «ValueArray не объявлено» при использовании функций карты искры в наборе данных, содержащем карты ha sh

Question

У меня есть класс

@Getter
@Setter
@NoArgsConstructor
public class TestInput implements Serializable {
    private Map<String, String> key1;
}

и json файл

[{"key1": {"key2": "key3"}}]

Я пытаюсь читать и работать с набором данных, используя функцию карты искры

Dataset<TestInput> input = sparkSession.read().json(inputPath).as(Encoders.bean(TestInput.class));

Dataset<TestInput> output = input.map((MapFunction<TestInput, TestInput>) x -> x, Encoders.bean(TestInput.class));

входные данные читаются правильно, так как я могу просмотреть ответ для input.show (), но ответ не включает фактическую карту

input.show()

+------+
|key1  |
+------+
|[key3]|
+------+

Но функция карты не работает (что также преобразует это в Java POJO) с ошибкой

A method named "valueArray" is not declared in any enclosing class nor any supertype, nor through a static import

Что я делаю не так?

Также, если я записываю ввод обратно в файл, я получаю обратно свой исходный json, поэтому он, вероятно, понимает карту, но не может преобразоваться в требуемый POJO

Answer 1

Если вы посмотрите на схему вашего input, вы увидите, что key1 - это не карта, а массив строк. Вот почему вы видите ошибку. Установите свою схему на карту, и вам должно быть хорошо go.

Scala код

val expectedSchema = StructType(Seq(StructField("key1", MapType(StringType, StringType))))

val input = spark.read
  .schema(expectedSchema)
  .json(path)
  .as(Encoders.bean(classOf[TestInput]))

// key1 should be of map type
input.printSchema()

val out = input.map(
  (x: TestInput) => {
    x.setKey1((Map(("newkey", "newval")) ++ x.getKey1.asScala).asJava)
    x
  },
  Encoders.bean(classOf[TestInput]),
)

input.show(truncate = false)
out.show(truncate = false)