嘿，这个问题我太熟了——本质上是Scala编译器在处理匿名BucketAssigner实现时的类型推断局限性导致的。

在Flink 1.10.0的Scala API里，StreamingFileSink.Builder的withBucketAssigner方法需要接收一个泛型参数明确的BucketAssigner[IN, BUCKET_ID]实例。但当你用匿名内部类去实现它的时候，Scala没法自动推导完整的泛型关联关系（尤其是你的UserEvent和String这两个泛型参数的绑定），结果就导致withBucketAssigner的返回类型直接退化成了Any，自然就找不到后续的.build()方法了。

1. 给forRowFormat显式指定泛型类型 #

最直接的方式就是在创建StreamingFileSink时，提前告诉编译器输入数据的类型，帮它锁定泛型信息：

val sink = StreamingFileSink.forRowFormat[UserEvent](new Path("s3a://123"), csvEncoder)
  .withRollingPolicy(
    DefaultRollingPolicy.builder()
      .withRolloverInterval(TimeUnit.MINUTES.toMinutes(5))
      .withInactivityInterval(TimeUnit.MINUTES.toMinutes(5))
      .withMaxPartSize(128 * 1024 * 1024)
      .build()
  )
  .withBucketAssigner(
    new BucketAssigner[UserEvent, String] {
      override def getBucketId(element: UserEvent, context: BucketAssigner.Context): String = element.getType.name
      override def getSerializer: SimpleVersionedSerializer[String] = new SimpleVersionedStringSerializer
    }
  )
  .build()

核心就是在forRowFormat后面加上[UserEvent]，让编译器提前知道输入类型，后续的方法调用就能正确推断返回类型了。

2. 把BucketAssigner改成独立命名类 #

把匿名类抽成一个单独的命名类，这样编译器能更清晰地识别泛型类型，代码也更易读复用：

// 单独定义BucketAssigner类
class UserEventBucketAssigner extends BucketAssigner[UserEvent, String] {
  override def getBucketId(element: UserEvent, context: BucketAssigner.Context): String = element.getType.name
  override def getSerializer: SimpleVersionedSerializer[String] = new SimpleVersionedStringSerializer
}

// 构建sink时直接实例化使用
val sink = StreamingFileSink
  .forRowFormat(new Path("s3a://123"), csvEncoder)
  .withRollingPolicy(
    DefaultRollingPolicy.builder()
      .withRolloverInterval(TimeUnit.MINUTES.toMinutes(5))
      .withInactivityInterval(TimeUnit.MINUTES.toMinutes(5))
      .withMaxPartSize(128 * 1024 * 1024)
      .build()
  )
  .withBucketAssigner(new UserEventBucketAssigner())
  .build()

Flink的Scala API在早期版本（比如1.10）的类型推断确实不如后续版本成熟，如果你有机会升级到1.13及以上版本，这类类型推断的坑会少很多。不过在当前1.10.0版本下，上面两种方法都能完美解决你的问题。

内容的提问来源于stack exchange，提问作者Patrick