StreamingFileSink with Custom Rolling Policy

streaming_file_sink_row_encode.scala

case class Writer(pathString: String, schema: String) {

  def build(): StreamingFileSink[GenericRecord] = {
    val path = new Path(pathString)
    val encoder = new ParquetEncoder();

    StreamingFileSink
      .forRowFormat(path, encoder)
      .withRollingPolicy(new MyRollingPolicy(60000))
      .withBucketAssigner(new MessageBucketAssigner)
      .build
  }

  class MyRollingPolicy(milliseconds: Long) extends RollingPolicy[GenericRecord, String] {
    var lastWindowEnd = Long.MinValue;

    override def shouldRollOnCheckpoint(partFileState: PartFileInfo[String]): Boolean = false

    override def shouldRollOnEvent(partFileState: PartFileInfo[String], element: GenericRecord): Boolean = {
      val logger_timestamp = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").parse(element.get("timestamp").asInstanceOf[String]).getTime()

      // Calculate Window
      val start = TimeWindow.getWindowStartWithOffset(logger_timestamp, 0, this.milliseconds);
      val end = start + this.milliseconds

      // Check if should break the file
      val output = lastWindowEnd != end && lastWindowEnd != Long.MinValue
      lastWindowEnd = end
      output
    }

    override def shouldRollOnProcessingTime(partFileState: PartFileInfo[String], currentTime: Long): Boolean = false
  }

  class ParquetEncoder extends Encoder[GenericRecord] {
    override def encode(element: GenericRecord, stream: OutputStream): Unit = {
      // TODO: Write to Parquet
      // TODO: Read from Parquet and write to OutputStream
    }
  }
}