如何在 Scala 中调用 FileIO.Write.via(Contextful, Contextful)

Question

我正在将 Apache Beam 与 Scala 结合使用。要为不同类型的流数据创建具有各自模式的新 ParquetIO.Sink 实例，我试图在 FileIO.Write 上调用 .via(Contextful, Contextful)。但是，IntelliJ 找不到 overloaded method that I'm trying to use 并引发此错误：Cannot resolve overloaded method 'via'.

FileIO
  .writeDynamic[String, DeserializedEvent]()
  .by(new UDFs.PartitionByEventName())
  .withDestinationCoder(StringUtf8Coder.of())
  .withNumShards(numShards)
  .withNaming(new UDFs.NameFiles())
  .via(
    Contextful.fn[DeserializedEvent, GenericRecord](
      new UDFs.EventToGenericRecord() // SerializableFunction[DeserializedEvent, String]
    ),
    Contextful.fn[String, ParquetIO.Sink](
      new UDFs.SinkParquet() // SerializableFunction[String, ParquetIO.Sink]
    )
  )
  .to(path)

这里有什么问题？

谢谢

Answer 1

Scala 的类型推断系统和重载解析系统不能完美地协同工作。在这种情况下，via() 函数的重载会阻止类型推断正常工作。对问题的解释很好

在您的情况下，您只需要将第二个 Contextful.fn 的类型参数更改为 [String, FileIO.Sink] 以使其与所需重载的签名相匹配。

如何在 Scala 中调用 FileIO.Write.via(Contextful, Contextful)

How to call FileIO.Write.via(Contextful, Contextful) in Scala

java

scala

parquet

apache-beam