“格式错误的数据长度为负数”，当尝试将来自 kafka 的 Spark 结构化流与 Avro 数据源一起使用时

Question

所以我一直在尝试使用 Kafka 和 Avro 数据 Structured-Streaming Avro 的 Angel Conde 结构化流媒体@

然而，我的数据似乎有点复杂，其中包含嵌套数据。这是我的代码，

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 中的任务 0.0 异常 (TID 8) org.apache.spark.SparkException: 无法执行用户 定义函数（$anonfun$27:（二进制）=> 结构，时间戳：字符串，sig_id：bigint，ip_type：字符串，dest_ip：字符串，优先级：bigint>） 在...

原因：com.twitter.bijection.InversionFailure：反转失败：[B@232f8415 at ...

原因：org.apache.avro.AvroRuntimeException：数据格式错误。长度为负：-25 at ...

我做错了吗？还是我的代码中的邪恶根源的嵌套模式？感谢你们的任何帮助

Answer 1

刚刚使用嵌套模式和使用新的 avro 数据源的示例更新了 repo。 Repo

在使用新数据源之前，我尝试使用双射库并遇到与您发布的相同的错误，但修复了它删除 Kafka 临时文件夹以重置旧排队数据。

最佳