我收到错误：使用 Kotlin 时 Apache Beam 中的 MapElements 转换出现

Question

我正在探索Github上的 GoogleCloudPlatform 提供的 Apache Beam 数据流模板。

特别是，我正在将PubSubToBigQuery模板从 Java 转换为 Kotlin。

通过这样做，我在在线转换中得到了一个Overload ambiguity resolution error 。错误信息是：MapElements.input(...).via(...)274

Error:(62, 22) Kotlin: Cannot choose among the following candidates without completing type inference: 

public final fun <NewInputT : Any!> via(fn: ((input: BigQueryInsertError!) -> FailsafeElement<String!, String!>!)!): MapElements<BigQueryInsertError!, FailsafeElement<String!, String!>!>! defined in org.apache.beam.sdk.transforms.MapElements

public final fun <NewInputT : Any!> via(fn: ((input: BigQueryInsertError!) -> FailsafeElement<String!, String!>!)!): MapElements<BigQueryInsertError!, FailsafeElement<String!, String!>!>! defined in org.apache.beam.sdk.transforms.MapElements

相关的 Java 代码片段是：

/*

     * Step 3 Contd.

     * Elements that failed inserts into BigQuery are extracted and converted to FailsafeElement

     */

    PCollection<FailsafeElement<String, String>> failedInserts =

        writeResult

            .getFailedInsertsWithErr()

            .apply(

                "WrapInsertionErrors",

                MapElements.into(FAILSAFE_ELEMENT_CODER.getEncodedTypeDescriptor())

                    .via((BigQueryInsertError e) -> wrapBigQueryInsertError(e)))

            .setCoder(FAILSAFE_ELEMENT_CODER);

Kotlin 转换如下所示：

/*

     * Step 3 Contd.

     * Elements that failed inserts into BigQuery are extracted and converted to FailsafeElement 

     */

val failedInserts: PCollection<FailsafeElement<String, String>> =

            writeResult.failedInsertsWithErr

            .apply(

                "WrapInsertionErrors",

                MapElements.into(FAILSAFE_ELEMENT_CODER.encodedTypeDescriptor)

                    .via { e: BigQueryInsertError -> wrapBigQueryInsertError(e) })

            .setCoder(FAILSAFE_ELEMENT_CODER)

我不知道如何解决这个问题。你能帮忙的话，我会很高兴。

Answer 1

原因是 Java 和 Kotlin 的重载规则略有不同，这意味着在 Kotlin 中有两个匹配的重载；public&nbsp;<NewInputT>&nbsp;MapElements<NewInputT,&nbsp;OutputT>&nbsp;via(ProcessFunction<NewInputT,&nbsp;OutputT>&nbsp;fn) public&nbsp;<NewInputT>&nbsp;MapElements<NewInputT,&nbsp;OutputT>&nbsp;via(SerializableFunction<NewInputT,&nbsp;OutputT>&nbsp;fn)最简单的解决方法是将 lambda 显式指定为 aSerializableFunction以获得正确的重载；.via<BigQueryInsertError>&nbsp;(SerializableFunction&nbsp;{&nbsp;e:&nbsp;BigQueryInsertError&nbsp;->&nbsp;wrapBigQueryInsertError(e)&nbsp;}))