Hi everyone, I’m using Flink with Java to read data from Kafka and write parquet into hdfs in streaming mode.

KafkaSource<KafkaMessage> source = KafkaSource.<KafkaMessage>builder()

.setBootstrapServers(brokers)

.setGroupId(groupId)

.setStartingOffsets(OffsetsInitializer.earliest())

.setTopics(topics)

.setDeserializer(new KafkaMessageDeserializer())

.build();

DataStream<SnowplowRawEvent> dataStreamSource =

env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source")

.map(kafkaMessage ->

SnowplowRawEvent.newBuilder()

.setKey(kafkaMessage.getKey())

.setEvent(kafkaMessage.getValue())

.setTimestamp(com.google.protobuf.Timestamp.newBuilder().build())

.setPartition(kafkaMessage.getPartition())

.setOffset(kafkaMessage.getOffset())

.setTopic(kafkaMessage.getTopic())

.build());

final FileSink<SnowplowRawEvent> sink = FileSink.forBulkFormat(

new Path(dataDirectory),

new ParquetWriterFactory<>((out) -> new ParquetProtoWriters.ParquetProtoWriterBuilder<>(out, SnowplowRawEvent.class)

.withCompressionCodec(CompressionCodecName.ZSTD)

.withBloomFilterEnabled(true)

.withDictionaryEncoding(true)

.build())

)

.withRollingPolicy(OnCheckpointRollingPolicy.build())

.withOutputFileConfig(new OutputFileConfig("part", ".zstd.parquet"))

.withNewBucketAssigner(new DateTimeBucketAssigner<>("yyyy-MM-dd"))

.build();

dataStreamSource.sinkTo(sink);

env.execute();

I am using com.google.protobuf to provide SnowplowRawEvent implementation of Message interface to further use it in constructor in

new ParquetProtoWriters.ParquetProtoWriterBuilder<>(out, SnowplowRawEvent.class)

. The problem comes when I try to write timestamps into hdfs, there seems to be only solution to provide timestamp in proto files which is importing it using

google.protobuf.Timestamp

, however when I read those parquets afterwards, those timestamp types turn into struct types with two fields seconds and nanos. I realize, that the problem is rather on google.protobuf side, but maybe someone has come across with the same problem. The generated SnowplowRawEvent class has timestamp with type com.google.protobuf.Timestamp