StreamingFileSink

streaming_file_sink.scala

val env = StreamExecutionEnvironment.getExecutionEnvironment

val source = new FlinkKafkaConsumer(settings.kafkaTopic(), new AugmentedMessageDeserializer, kafkaProperties)
val writer = Writer(settings.s3Path(), GenericRecordSchema.schema.toString()).build()

env
  .enableCheckpointing(checkpointInterval)
  .addSource(source)
  .addSink(writer)

env.execute()

// Writer.scala

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

  def build(): StreamingFileSink[GenericRecord] = {
    val path = new Path(pathString)
    val builder: ParquetBuilder[GenericRecord] = (out: OutputFile) =>
      createAvroParquetWriter(schema, GenericData.get, out)
    val writer = new ParquetWriterFactory(builder)

    StreamingFileSink
      .forBulkFormat(path, writer)
      .withBucketAssigner(new MessageBucketAssigner)
      .build
  }

  @throws[IOException]
  private def createAvroParquetWriter[T](schemaString: String, dataModel: GenericData, out: OutputFile) = {
    val schema = new Schema.Parser().parse(schemaString)

    AvroParquetWriter
      .builder[T](out)
      .withSchema(schema)
      .withDataModel(dataModel)
      .withWriteMode(ParquetFileWriter.Mode.OVERWRITE)
      .build
  }
}

Hi Can u please help me with AugmentedMessageDeserializer and GenericRecordSchema class code.

What do you need help with? https://gist.github.com/speeddragon/06102e8f3e1f2154dc3b69f072132cd1 **GenericRecordSchema** have the following code: ```scala import org.apache.avro.SchemaBuilder object GenericRecordSchema { val schema = SchemaBuilder .record(RECORD.name) .fields() .requiredString(KEY.name) .name(MESSAGE.name) .`type`() .bytesType() .noDefault() .requiredString(HEADERS.name) .requiredString(TIMESTAMP.name) .endRecord() } ```