Как создать набор данных в Apache Spark со сложной схемой ⇐ JAVA

[Anonymous]

У меня есть Java-приложение Spark, в котором мне нужно создать объект набора данных. Хитрость в том, что он имеет несколько слоев, что создает проблемы при использовании метода createDataFrame(...). (Фиктивную) схему для моего набора данных можно определить следующим образом:

Код: Выделить всё

        StructType objectEntities = new StructType(new StructField[]{
new StructField("id",DataTypes.StringType,true, Metadata.empty()),
new StructField("type",DataTypes.StringType,true,Metadata.empty())
});
StructType subjectEntities = new StructType(new StructField[]{
new StructField("id",DataTypes.StringType,true, Metadata.empty()),
new StructField("type",DataTypes.StringType,true,Metadata.empty())
});

StructType innerLevelStruct = new StructType();
innerLevelStruct = innerLevelStruct.add("f1",DataTypes.StringType,false);
innerLevelStruct = innerLevelStruct.add("f2",DataTypes.StringType,false);
innerLevelStruct = innerLevelStruct.add("f3",DataTypes.StringType,false);
innerLevelStruct = innerLevelStruct.add("f4",DataTypes.StringType,false);
innerLevelStruct = innerLevelStruct.add("f5",objectEntities,true);
innerLevelStruct = innerLevelStruct.add("f6",subjectEntities,true);

StructType outerLayerStruct = new StructType();
outerLayerStruct = outerLayerStruct.add("id",DataTypes.StringType,false);
outerLayerStruct = outerLayerStruct.add("type",DataTypes.StringType,false);
outerLayerStruct = outerLayerStruct.add("state",DataTypes.StringType,false);
outerLayerStruct = outerLayerStruct.add("items",innerLevelStruct,false);

StructType entityDetailsSchema = new StructType();
entityDetailsSchema = entityDetailsSchema.add("id",DataTypes.StringType,false);
entityDetailsSchema = entityDetailsSchema.add("cols",outerLayerStruct,false);

Поскольку схема определена, я подумал, что могу определить данные таким образом.

Код: Выделить всё

        List entries = new ArrayList();
entries.add(RowFactory.create("id",List.of(
"testId","testType","testState",
List.of(
"f1Val","f2Val","f3Val","f4Val",List.of("a","b"),List.of("c","v")
)
)));
Dataset output = sparkSession.createDataFrame(entries,entityDetailsSchema);

Где sparkSession — это стандартные сеансы Spark, определенные следующим образом:

Код: Выделить всё

sparkSession = SparkSession.builder()
.master("local[1]")
.appName("TestSpark")
.getOrCreate();

Обычно это работает для наборов данных с простой одноуровневой схемой. Однако когда я запускаю свое приложение со схемой, упомянутой в начале, я получаю:

Код: Выделить всё

java.lang.IllegalArgumentException: The value ([testId, testType, testState, [f1Val, f2Val, f3Val, f4Val, [a, b], [c, v]]]) of the type (java.util.ImmutableCollections.ListN) cannot be converted to struct

Я подозреваю, что это потому, что я использую List.of(...). Проблема устраняется путем замены List.of(...) на RowFactory.create(...), но это не подходящее решение, поскольку оно создает проблемы, когда я пытаюсь преобразовать набор данных в POJO через кодировщик следующим образом:

Код: Выделить всё

output.as(Encoders.bean(MyPojoClass.class))
.foreachPartition(partition -> {
// do work
});

Выдает мне исключение, сообщающее, что оно ожидает массивы вместо structType:

Код: Выделить всё

org.apache.spark.sql.AnalysisException: [UNSUPPORTED_DESERIALIZER.DATA_TYPE_MISMATCH] The deserializer is not supported: need a(n) "ARRAY" field but got "STRUCT

Итак, вопрос заключается в следующем: как я могу создать набор данных, используя предоставленную схему? Как я могу определить строку, которая будет заполнять каждый слой данными? Способен ли RowFactory на это вообще - судя по тому, что я читал, это линейный простой метод, поэтому не уверен, что его вообще правильно использовать?

Подробнее здесь: https://stackoverflow.com/questions/787 ... lex-schema