Flink 迁移：BucketingSink[T] 到 StreamingFileSink[T]答案

【问题标题】：Flink migration: BucketingSink[T] to StreamingFileSink[T]Flink 迁移：BucketingSink[T] 到 StreamingFileSink[T]
【发布时间】：2020-10-28 16:39:11
【问题描述】：

我有最旧的 Flink 版本，想更新到最新的稳定版本。 BucketingSink 在最新版本中被弃用，我尝试用 StreamingFileSink 更改它。要初始化它，我使用 StreamingFileSink.forBulkFormat 但有错误：

type arguments [T] do not conform to method forSpecificRecord's type parameter bo
unds [T <: org.apache.avro.specific.SpecificRecordBase]
[ERROR] .forBulkFormat(new Path(s"${conf.output}/$path") , AvroWriters.forSpecificRecord[T](classOf[T]) )

我也找不到如何设置bucketer:DateTimeBucketer[T], inactiveBucketThreshold, writer: Writer[T]

你能帮我找到正确的方法吗？

旧代码：

trait Runner[T <: SpecificRecordBase] extends Serializable {
      def createHdfsSink(conf: FlinkConfig, path: String): BucketingSink[T] = {
            val bucketer = new DateTimeBucketer[T]
            val sink = new BucketingSink[T](s"${conf.output}/$path")
            sink
              .setBatchSize(toBytes(conf.batchSize))
              .setBucketer(bucketer)
              .setInactiveBucketThreshold(toMillis(conf.inactiveBucketThreshold))
              .setWriter(writer)
              .setPendingPrefix(pendingPrefix)
              .setBatchRolloverInterval(conf.fileOpenIntervalTime)
      }

有错误的新代码：

  def createHdfsStreamingSink[T : ClassTag](conf: FlinkConfig, path: String): StreamingFileSink[T] = {
    val sink = StreamingFileSink
      .forBulkFormat(new Path(s"${conf.output}/$path") , AvroWriters.forSpecificRecord[T](classOf[T]) )
      .build()

    // TODO:        .withOutputFileConfig()
    sink
  }

【问题讨论】：

标签： java scala apache-flink

【解决方案1】：

我认为，您应该改用forReflectRecord(Class<T> type) 方法，该方法将使用反射为该类型创建架构并使用该架构来写入记录。自定义存储桶分配器在StreamingFileSink 设置期间配置，并通过.withBucketAssigner(BucketAssigner<IN, ID> assigner) 方法调用指定。

所以，最后你的StreamingFileSink builder 将如下所示：

def createHdfsStreamingSink[T : ClassTag](conf: FlinkConfig, path: String): StreamingFileSink[T] = {
    val sink = StreamingFileSink
      .forBulkFormat(new Path(s"${conf.output}/$path") , AvroWriters.forReflectRecord[T](classOf[T]) )
      .withBucketAssigner(bucketAssigner)
      .build()
    sink
}

【讨论】：