Hi,
I am trying to upload avro records to AWS S3 using StreamingFileSink.
Avro file is getting created and uploaded with valid data but I want to add a
Rolling policy which will roll the file after specific time or total part file
size.
With forBulkFormat() I am able to use only CheckpointRollingPolicy which is
automatically rolling all part files on every check point.
Though CheckpointRollingPolicy allows us to override 'shouldRollOnEvent' &
'shouldRollOnProcessingTime' it is not behaving as per overrided logic and
just rolls out on each check point.
-'shouldRollOnProcessingTime' method is not getting invoked for each streaming
message, its
- 'shouldRollOnEvent' part file size here always shows fixed single message
size only not the clubbed part file size
Below is the code snippet -
val avroOcfFilesink : StreamingFileSink[GenericRecord] =
StreamingFileSink.forBulkFormat(new Path(avroOutputPath),
new AvroWriterFactory[GenericRecord](new AvroBuilder[GenericRecord]() {
override def createWriter(out: OutputStream):
DataFileWriter[GenericRecord] = {
val schema: Schema = new
Schema.Parser().parse(faultCodeOCFRecordSchema)
val datumWriter = new ReflectDatumWriter[GenericRecord](schema)
val dataFileWriter = new DataFileWriter[GenericRecord](datumWriter)
//dataFileWriter.setCodec(CodecFactory.snappyCodec)
dataFileWriter.create(schema, out)
dataFileWriter
}
}))
.withBucketAssigner(new BucketAssigner[GenericRecord, String] {
override def getBucketId(in: GenericRecord, context: Context): String =
{
val bucketIdPrefix =
configurationParameters.getRequired("s3.bucket.id.prefix")
val currentProcessingTimeUTC = System.currentTimeMillis()
bucketIdPrefix +
TimeConversion.convertTimestampToRunDate_HHMM(currentProcessingTimeUTC)
}
override def getSerializer: SimpleVersionedSerializer[String] = {
SimpleVersionedStringSerializer.INSTANCE }
}).
withRollingPolicy(
new CheckpointRollingPolicy[GenericRecord, String] {
override def shouldRollOnEvent(partFileState: PartFileInfo[String],
element: GenericRecord): Boolean = {
println("partFileState.getSize:"+partFileState.getSize)
(partFileState.getSize >= 1024*8)
}
override def shouldRollOnProcessingTime(partFileState:
PartFileInfo[String], currentTime: Long): Boolean = {
println("currentTime:"+currentTime+" ,
partFileState.getCreationTime"+partFileState.getCreationTime+",
Diff:"+(currentTime - partFileState.getCreationTime))
(currentTime - partFileState.getCreationTime >= 600000)
}
}
).build()
avroRecordStream.addSink(avroOcfFilesink).setParallelism(1).name("AvroToS3Bucket")
