org.apache.kafka.connect.errors.DataException：结构模式不匹配

Question

"airingTo" : [ {
      "dow" : [ "SATURDAY" ]
      }]

public static Struct createAiringStruct() 
{
    Schema valueSchema = createValueSchemaForProgram();
    Struct valueStruct = new Struct(valueSchema).put("airingTo", 
    getAiringList());
    return valueStruct;
}

private static Schema createValueSchemaForProgram() 
{
    return 
    SchemaBuilder.struct().name("PROGRAM").field("airingTo", 
    SchemaBuilder.array(createAiringListSchema())).build();
}

private static Schema createAiringListSchema() 
{
   return SchemaBuilder.struct().name("airingTo").field("dow", 
   SchemaBuilder.array(Schema.STRING_SCHEMA)).build();
}

private static List<Struct> getAiringList() 
{
    Struct valueStruct = new 
    Struct(createAiringListSchema()).put("dow", 
    Arrays.asList("SATURDAY"));
    List<Struct> dayPartList = new ArrayList<>();
    dayPartList.add(valueStruct);
    return dayPartList;
}

Answer 1

不确定从哪里得到.name（“program”），但这似乎对我有用。

看来你把结构名称和字段混在一起了。

    Schema dowSchema = SchemaBuilder.array(Schema.STRING_SCHEMA).build();
    Schema airingToItemSchema = SchemaBuilder.struct()
            .field("dow", dowSchema)
            .build();
    Schema airingToSchema = SchemaBuilder.array(airingToItemSchema).build();
    Schema rootSchema = SchemaBuilder.struct()
            .field("airingTo", airingToSchema);
            .build();

    Struct item = new Struct(airingToItemSchema)
            .put("dow", Collections.singletonList("SATURDAY"));
    Struct rootStruct = new Struct(rootSchema)
            .put("airingTo", Collections.singletonList(item));

    System.out.println(rootStruct);

产出