所以我一直在尝试angel conde的结构化流媒体，使用kafka和avro数据结构化流媒体avro
然而，我的数据似乎有点复杂，因为其中有嵌套的数据。这是我的密码，

private static Injection<GenericRecord, byte[]> recordInjection;
private static StructType type;
private static final String SNOQTT_SCHEMA = "{"
        +"\"type\": \"record\","
        +"\"name\": \"snoqttv2\","
        +"\"fields\": ["
        +"    { \"name\": \"src_ip\", \"type\": \"string\" },"
        +"    { \"name\": \"classification\", \"type\": \"long\" },"
        +"    { \"name\": \"device_id\", \"type\": \"string\" },"
        +"    { \"name\": \"alert_msg\", \"type\": \"string\" },"
        +"    { \"name\": \"src_mac\", \"type\": \"string\" },"
        +"    { \"name\": \"sig_rev\", \"type\": \"long\" },"
        +"    { \"name\": \"sig_gen\", \"type\": \"long\" },"
        +"    { \"name\": \"dest_mac\", \"type\": \"string\" },"
        +"    { \"name\": \"packet_info\", \"type\": {"
        +"        \"type\": \"record\","
        +"        \"name\": \"packet_info\","
        +"        \"fields\": ["
        +"              { \"name\": \"DF\", \"type\": \"boolean\" },"
        +"              { \"name\": \"MF\", \"type\": \"boolean\" },"
        +"              { \"name\": \"ttl\", \"type\": \"long\" },"
        +"              { \"name\": \"len\", \"type\": \"long\" },"
        +"              { \"name\": \"offset\", \"type\": \"long\" }"
        +"          ],"
        +"        \"connect.name\": \"packet_info\" }},"
        +"    { \"name\": \"timestamp\", \"type\": \"string\" },"
        +"    { \"name\": \"sig_id\", \"type\": \"long\" },"
        +"    { \"name\": \"ip_type\", \"type\": \"string\" },"
        +"    { \"name\": \"dest_ip\", \"type\": \"string\" },"
        +"    { \"name\": \"priority\", \"type\": \"long\" }"
        +"],"
        +"\"connect.name\": \"snoqttv2\" }";

private static Schema.Parser parser = new Schema.Parser();
private static Schema schema = parser.parse(SNOQTT_SCHEMA);

static {
    recordInjection = GenericAvroCodecs.toBinary(schema);
    type = (StructType) SchemaConverters.toSqlType(schema).dataType();
}

public static void main(String[] args) throws StreamingQueryException{
    // Set log4j untuk development langsung dari java
    LogManager.getLogger("org.apache.spark").setLevel(Level.WARN);
    LogManager.getLogger("akka").setLevel(Level.ERROR);

    // Set konfigurasi untuk streamcontext dan sparkcontext
    SparkConf conf = new SparkConf()
            .setAppName("Snoqtt-Avro-Structured")
            .setMaster("local[*]");

    // Inisialisasi spark session
    SparkSession sparkSession = SparkSession
            .builder()
            .config(conf)
            .getOrCreate();

    // Reduce task number
    sparkSession.sqlContext().setConf("spark.sql.shuffle.partitions", "3");

    // Mulai data stream di kafka
    Dataset<Row> ds1 = sparkSession
            .readStream()
            .format("kafka")
            .option("kafka.bootstrap.servers", "localhost:9092")
            .option("subscribe", "snoqttv2")
            .option("startingOffsets", "latest")
            .load();

    // Mulai streaming query

    sparkSession.udf().register("deserialize", (byte[] data) -> {
        GenericRecord record = recordInjection.invert(data).get();
        return RowFactory.create(
                record.get("timestamp").toString(),
                record.get("device_id").toString(),
                record.get("ip_type").toString(),
                record.get("src_ip").toString(),
                record.get("dest_ip").toString(),
                record.get("src_mac").toString(),
                record.get("dest_mac").toString(),
                record.get("alert_msg").toString(),
                record.get("sig_rev").toString(),
                record.get("sig_gen").toString(),
                record.get("sig_id").toString(),
                record.get("classification").toString(),
                record.get("priority").toString());
    }, DataTypes.createStructType(type.fields()));

    ds1.printSchema();
    Dataset<Row> ds2 = ds1
            .select("value").as(Encoders.BINARY())
            .selectExpr("deserialize(value) as rows")
            .select("rows.*");

    ds2.printSchema();

    StreamingQuery query1 = ds2
            .groupBy("sig_id")
            .count()
            .writeStream()
            .queryName("Signature ID Count Query")
            .outputMode("complete")
            .format("console")
            .start();

    query1.awaitTermination();
}

这一切都很有趣和游戏，直到我得到第一批信息，它遇到了错误
18/01/22 14:29:00错误执行器：阶段4.0（tid 8）中任务0.0出现异常org.apache.spark.sparkexception:未能执行用户定义函数（$anonfun$27:（binary）=>struct，timestamp:string，sig\u id:bigint，ip地址_type:string，dest\ u ip:字符串，priority:bigint>)在。。。
原因：com.twitter.bijection.inversionfailure:无法反转：[b@232f8415 在。。。
原因：org.apache.avro.avroruntimeexception:数据格式错误。长度为负：-25在。。。
我做错了吗？或者我的嵌套模式是我代码中邪恶的根源？感谢你们的帮助