I am using following method to write data in parquet file :
static void toParquet() {
String schema = "message spark_schema {\n optional binary stringField (UTF8);\n optional INT96 createdAt;\n}";
MessageType readSchema = MessageTypeParser.parseMessageType(schema);
Configuration configuration = new Configuration();
GroupWriteSupport.setSchema(readSchema, configuration);
SimpleGroupFactory sfg = new SimpleGroupFactory(readSchema);
Path file = new Path("/home/user/data-" + System.currentTimeMillis() + ".parquet");
try {
ParquetWriter<Group> writer = new ParquetWriter<Group>(file, new GroupWriteSupport(), CompressionCodecName.UNCOMPRESSED, 1024, 1024, 512,
true, false, ParquetProperties.WriterVersion.PARQUET_1_0, configuration);
for (int i = 0; i < 10000; ++i) {
writer.write(sfg.newGroup().append("stringField", "abc").append("createdAt", String.valueOf(System.currentTimeMillis())));
}
writer.close();
} catch (Exception e) {
e.printStackTrace();
}
}
But it gives me following exception :
java.lang.IllegalArgumentException: Fixed Binary size 13 does not match field type length 12 at org.apache.parquet.column.values.plain.FixedLenByteArrayPlainValuesWriter.writeBytes(FixedLenByteArrayPlainValuesWriter.java:53) at org.apache.parquet.column.values.dictionary.DictionaryValuesWriter$PlainFixedLenArrayDictionaryValuesWriter.createDictionaryPage(DictionaryValuesWriter.java:324) at org.apache.parquet.column.values.fallback.FallbackValuesWriter.createDictionaryPage(FallbackValuesWriter.java:102) at org.apache.parquet.column.impl.ColumnWriterV1.flush(ColumnWriterV1.java:242) at org.apache.parquet.column.impl.ColumnWriteStoreV1.flush(ColumnWriteStoreV1.java:126) at org.apache.parquet.hadoop.InternalParquetRecordWriter.flushRowGroupToStore(InternalParquetRecordWriter.java:164) at org.apache.parquet.hadoop.InternalParquetRecordWriter.checkBlockSizeReached(InternalParquetRecordWriter.java:141) at org.apache.parquet.hadoop.InternalParquetRecordWriter.write(InternalParquetRecordWriter.java:123) at org.apache.parquet.hadoop.ParquetWriter.write(ParquetWriter.java:288) at com.app.test.SimpleParquet.toParquet(SimpleParquet.java:498) at com.app.test.SimpleParquet.main(SimpleParquet.java:63)
As System.currentTimeMillis() gives 13 byte value and epoc time is of 10 byte how to provide int96 value for current timestamp which is a fixed length 12 byte array.
I found some threads in which they say that the timestamp is combination of julian day and time in nanosecs, how can i convert current timestamp into this format?
Following piece of code worked for me in which i have calculated NanoTime and then called toBinary function on it.
While writing in parquet used above method -