Thanks a lot Ryan for the clarified response!!! :) I will surely figure out the steps to start contributing :) And will try out parquet-avro.. :)
-----Original Message----- From: Ryan Blue [mailto:[email protected]] Sent: Wednesday, November 04, 2015 10:37 PM To: Manisha Sethi <[email protected]>; [email protected] Subject: Re: Timestamp and time not being written Unsopported The problem you're hitting is that Hive doesn't support the int64 timestamps that are in the Parquet spec. It supports an undocumented timestamp format instead and we need to get it moved over to using the int64 timestamps. If you want to continue using Hive's Writable objects in your program and use the int64 timestamp, then I recommend contributing support to Hive. It doesn't look like it would be that hard, since you already understand the code path and where this is going wrong. I don't know if I would recommend Hive as an object model. I usually recommend parquet-avro, since that enables you to use the same object model with both Avro and Parquet file formats. rb On 11/03/2015 07:22 PM, Manisha Sethi wrote: > Thanks Ryan!! > > > I am trying to write in parquet format using hive-exec1.2.1 , this jar > supports to write to ORC/Parquet file formats hive supports. > My sample code looks like : > > MessageType schema = MessageTypeParser.parseMessageType("message > basket { required int64 time (TIMESTAMP_MILLIS); }"); > Configuration config = new Configuration(); > config.set("fs.default.name", "hdfs://<ip>:9000"); > String outFilePath; > Path outDirPath = new > Path("hdfs://<ip>:9000/user/hdfs/test5"); > > ParquetWriter writer = new ParquetWriter(outDirPath, new > DataWritableWriteSupport() > { > private DataWritableWriter writer; > > @Override > public WriteContext init(Configuration configuration) > { > if > (configuration.get(DataWritableWriteSupport.PARQUET_HIVE_SCHEMA) == null) > { > > configuration.set(DataWritableWriteSupport.PARQUET_HIVE_SCHEMA, > schema.toString()); > } > return super.init(configuration); > } > > }, CompressionCodecName.SNAPPY, 256 * 1024 * 1024, 100 * > 1024, 100 * 1024, true, false, WriterVersion.PARQUET_2_0, config); > > List<ObjectInspector> list = new ArrayList<ObjectInspector>(); > ObjectInspector tins = > ObjectInspectorFactory.getReflectionObjectInspector(Timestamp.class, > ObjectInspectorOptions.JAVA); > list.add(tins); > > List<String> columnnames = new ArrayList<String>(); > columnnames.add("time"); > > StructObjectInspector inspec = > ObjectInspectorFactory.getStandardStructObjectInspector(columnnames, > list); > > List<Object> obj = new ArrayList<Object>(); > obj.add(new Timestamp(new Date().getTime())); > writer.write(new ParquetHiveRecord(obj, inspec)); > writer.close(); > > ==================================================================== > > As timestamp is defined in parquet schema by annotating an "INT64"(which > designates a long) so it instantiates a LongStatistics object used after > writing values. > > Using code like(parquet.column.statistics.Statistics.java): > public static Statistics getStatsBasedOnType(PrimitiveType.PrimitiveTypeName > type) > /* */ { > /* 48 */ switch > (1.$SwitchMap$parquet$schema$PrimitiveType$PrimitiveTypeName[type.ordinal()]) > { > /* */ case 1: > /* 50 */ return new IntStatistics(); > /* */ case 2: > /* 52 */ return new LongStatistics(); > /* */ case 3: > /* 54 */ return new FloatStatistics(); > /* */ case 4: > /* 56 */ return new DoubleStatistics(); > /* */ case 5: > /* 58 */ return new BooleanStatistics(); > /* */ case 6: > /* 60 */ return new BinaryStatistics(); > /* */ case 7: > /* 62 */ return new BinaryStatistics(); > /* */ case 8: > /* 64 */ return new BinaryStatistics(); > /* */ } > /* 66 */ throw new UnknownColumnTypeException(type); > /* */ } > > > But when Timestamp is written, below code is executed and written in binary > form: > org.apache.hadoop.hive.ql.io.parquet.write.DataWritableWriter.java > > Line 295: case TIMESTAMP: > Timestamp ts = ((TimestampObjectInspector) > inspector).getPrimitiveJavaObject(value); > recordConsumer.addBinary(NanoTimeUtils.getNanoTime(ts, > false).toBinary()); > break; > > Which leads to : > Parquet.Column.Impl.ColumnWriterV2.java > public void write(Binary value, int repetitionLevel, int definitionLevel) > /* */ { > /* 154 */ if (DEBUG) log(value, repetitionLevel, definitionLevel); > /* 155 */ repetitionLevel(repetitionLevel); > /* 156 */ definitionLevel(definitionLevel); > /* 157 */ this.dataColumn.writeBytes(value); > /* 158 */ this.statistics.updateStats(value);====>>>>FAILS > /* 159 */ this.valueCount += 1; > /* */ } > > Since statistics is LongStatistics instantiated and it has method defined > which accepts long value, but call is made with binary argument, and call > goes to Base Class Statistics.java where it throws unsupported exception. > > Complete StackTrace: > > Exception in thread "main" java.lang.RuntimeException: Parquet record is > malformed: null > at > org.apache.hadoop.hive.ql.io.parquet.write.DataWritableWriter.write(DataWritableWriter.java:64) > at > org.apache.hadoop.hive.ql.io.parquet.write.DataWritableWriteSupport.write(DataWritableWriteSupport.java:59) > at > org.apache.hadoop.hive.ql.io.parquet.write.DataWritableWriteSupport.write(DataWritableWriteSupport.java:31) > at > parquet.hadoop.InternalParquetRecordWriter.write(InternalParquetRecordWriter.java:121) > at parquet.hadoop.ParquetWriter.write(ParquetWriter.java:258) > at ParquetTestWriter$1.run(ParquetTestWriter.java:126) > at ParquetTestWriter$1.run(ParquetTestWriter.java:1) > at java.security.AccessController.doPrivileged(Native Method) > at javax.security.auth.Subject.doAs(Subject.java:422) > at > org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1628) > at ParquetTestWriter.main(ParquetTestWriter.java:37) > Caused by: java.lang.UnsupportedOperationException > at > parquet.column.statistics.Statistics.updateStats(Statistics.java:115) > at parquet.column.impl.ColumnWriterV2.write(ColumnWriterV2.java:158) > at > parquet.io.MessageColumnIO$MessageColumnIORecordConsumer.addBinary(MessageColumnIO.java:346) > at > org.apache.hadoop.hive.ql.io.parquet.write.DataWritableWriter.writePrimitive(DataWritableWriter.java:297) > at > org.apache.hadoop.hive.ql.io.parquet.write.DataWritableWriter.writeValue(DataWritableWriter.java:106) > at > org.apache.hadoop.hive.ql.io.parquet.write.DataWritableWriter.writeGroupFields(DataWritableWriter.java:89) > at > org.apache.hadoop.hive.ql.io.parquet.write.DataWritableWriter.write(DataWritableWriter.java:60) > ... 10 more > > Appreciate your help! > > -Manisha > > -----Original Message----- > From: Ryan Blue [mailto:[email protected]] > Sent: Tuesday, November 03, 2015 10:19 PM > To: [email protected] > Subject: Re: Timestamp and time not being written Unsopported > > On 11/03/2015 12:35 AM, Manisha Sethi wrote: >> Hi >> >> I am trying to write timestamp using int64 (TIMESTAMP_MILLIS) via >> ParquetWriter using jar hive-exec 1.2.1... But getting unsopprted >> exception... >> Issue is : when call reaches "add binary" >> >> break; >> case BINARY: >> byte[] vBinary = ((BinaryObjectInspector) >> inspector).getPrimitiveJavaObject(value); >> recordConsumer.addBinary(Binary.fromByteArray(vBinary)); >> break; >> case TIMESTAMP: >> Timestamp ts = ((TimestampObjectInspector) >> inspector).getPrimitiveJavaObject(value); >> recordConsumer.addBinary(NanoTimeUtils.getNanoTime(ts, >> false).toBinary()); >> break; >> case DECIMAL: >> HiveDecimal vDecimal = >> ((HiveDecimal)inspector.getPrimitiveJavaObject(value)); >> DecimalTypeInfo decTypeInfo = >> (DecimalTypeInfo)inspector.getTypeInfo(); >> recordConsumer.addBinary(decimalToBinary(vDecimal, decTypeInfo)); >> break; >> >> Then in Columnwriter it fails at updatestatistics, since call is made >> using longstatistic(corrs to its int64 data type but value is binary >> which is not defined) >> >> this.repetitionLevelColumn.writeInteger(repetitionLevel); >> /* 203 */ this.definitionLevelColumn.writeInteger(definitionLevel); >> /* 204 */ this.dataColumn.writeBytes(value); >> /* 205 */ updateStatistics(value); >> >> this.statistics.updateStats(value);====>>>> Method is not defined for >> LongStatistics, hence throws unsupported exception >> >> ________________________________ >> >> > > Manisha, > > Thanks for taking the time to e-mail about this. How are you writing this > timestamp? Are you using Hive, or are you trying to use Hive's object model > in your own code? > > Could you also send the stack trace that you're seeing? I'm confused about > why the method would be undefined, since it should be defined for the types > correctly. > > Thanks, > > rb > > -- > Ryan Blue > Software Engineer > Cloudera, Inc. > > ________________________________ > -- Ryan Blue Software Engineer Cloudera, Inc. ________________________________
