java - 如何使用java将timestamp-millis logicalType序列化为avro文件

Question

我有一个场景，我想创建具有时间戳列的 avro 文件，它看起来像 2016-11-16 06:43:19.77

我使用avro-1.8.2.jar 编写avro 文件。这对于原始类型的记录非常有效，但对于像时间戳 - 毫秒这样的逻辑类型面临的问题。

架构:

{
  "type": "record",
  "name": "MyRecord",
  "namespace": "org.demo",
  "fields": [
    {
      "name": "timestamp_with_logical_type",
      "type": {
        "type": "long",
        "logicalType": "timestamp-millis"
      }
    },
    {
      "name": "timestamp_no_logical_type",
      "type": "long"
    }
  ]
}

Java 代码:

       //Instantiating the GenericRecord class.
        GenericRecord record  = new Record(schema);
        long millis = Instant.now().toEpochMilli();
        //Insert data according to schema
        record.put("timestamp_with_logical_type", new Timestamp(millis));
        record.put("timestamp_no_logical_type", millis);

        DataFileWriter<GenericRecord> dataFileWriter = null;
        DatumWriter<GenericRecord> datumWriter = new GenericDatumWriter<GenericRecord>(schema);
        dataFileWriter = new DataFileWriter<GenericRecord>(datumWriter);
        dataFileWriter.setCodec(CodecFactory.snappyCodec());
        dataFileWriter.setFlushOnEveryBlock(true);
        dataFileWriter.setSyncInterval(32);
        dataFileWriter.create(SCHEMA,new File("E:\\MyFiles\\Avro_Data\\demo.avro"));
        dataFileWriter.append(record);
        dataFileWriter.close();

上面的代码给出了以下异常:

Exception in thread "main" org.apache.avro.file.DataFileWriter$AppendWriteException: java.lang.ClassCastException: java.sql.Timestamp cannot be cast to java.lang.CharSequence
    at org.apache.avro.file.DataFileWriter.append(DataFileWriter.java:308)
    at AvroFileReaderWriter.writeToAvro(AvroFileReaderWriter.java:264)
    at AvroFileReaderWriter.main(AvroFileReaderWriter.java:74)
Caused by: java.lang.ClassCastException: java.sql.Timestamp cannot be cast to java.lang.CharSequence
    at org.apache.avro.generic.GenericDatumWriter.writeString(GenericDatumWriter.java:267)
    at org.apache.avro.generic.GenericDatumWriter.writeString(GenericDatumWriter.java:262)
    at org.apache.avro.generic.GenericDatumWriter.writeWithoutConversion(GenericDatumWriter.java:128)
    at org.apache.avro.generic.GenericDatumWriter.write(GenericDatumWriter.java:75)
    at org.apache.avro.generic.GenericDatumWriter.writeField(GenericDatumWriter.java:166)
    at org.apache.avro.generic.GenericDatumWriter.writeRecord(GenericDatumWriter.java:156)
    at org.apache.avro.generic.GenericDatumWriter.writeWithoutConversion(GenericDatumWriter.java:118)
    at org.apache.avro.generic.GenericDatumWriter.write(GenericDatumWriter.java:75)
    at org.apache.avro.generic.GenericDatumWriter.write(GenericDatumWriter.java:62)
    at org.apache.avro.file.DataFileWriter.append(DataFileWriter.java:302)
    ... 2 more

Answer 1

我认为你需要使用 Instant 而不是 Timestamp，所以在 java 代码中:

Java 代码:

   //Instantiating the GenericRecord class.
    GenericRecord record  = new Record(schema);
    Instant millis = Instant.now();
    //Insert data according to schema
    record.put("timestamp_with_logical_type", millis));        
    record.put("timestamp_no_logical_type", millis.toEpochMilli());

    DataFileWriter<GenericRecord> dataFileWriter = null;
    DatumWriter<GenericRecord> datumWriter = new GenericDatumWriter<GenericRecord>(schema);
    dataFileWriter = new DataFileWriter<GenericRecord>(datumWriter);
    dataFileWriter.setCodec(CodecFactory.snappyCodec());
    dataFileWriter.setFlushOnEveryBlock(true);
    dataFileWriter.setSyncInterval(32);
    dataFileWriter.create(SCHEMA,new File("E:\\MyFiles\\Avro_Data\\demo.avro"));
    dataFileWriter.append(record);
    dataFileWriter.close();