我已經檢查了這個類似但有 7 年歷史的問題，但它不適用于較新的 Flink 版本。
我正在嘗試運行一個簡單的 Flink Kafka 作業，并嘗試了各種版本，每個版本都有不同的編譯錯誤。我正在使用 sbt 來管理我的依賴項：
val flinkDependencies = Seq(
"org.apache.flink" %% "flink-clients" % flinkVersion,
"org.apache.flink" %% "flink-scala" % flinkVersion,
"org.apache.flink" %% "flink-streaming-scala" % flinkVersion,
"org.apache.flink" %% "flink-connector-kafka" % flinkVersion
)
嘗試的版本：
scala 2.11.12 和 2.12.15
flink 1.14.6
我正在嘗試編譯的代碼（相關位）：
import org.apache.flink.streaming.util.serialization.SimpleStringSchema
import org.apache.flink.connector.kafka.source.KafkaSource
...
val env = ExecutionEnvironment.getExecutionEnvironment
val kafkaConsumer = new KafkaSource.builder[String]
.setBootstrapservers("localhost:9092")
.setGroupId("flink")
.setTopics("test")
.build()
val text = env.fromSource(kafkaConsumer)
我沒有找到一個官方的例子，這確實是一個人應該如何使用，KafkaSource但我在這里和這里找到了這個設定。在我仍然非常新的 Java 眼中，這看起來與API 檔案一致。但是是的，它不能與任何一個 Scala 版本一起作業：
[error] somepathwithmyfile: type builder is not a member of object org.apache.flink.connector.kafka.source.KafkaSource
[error] val kafkaConsumer = new KafkaSource.builder[String]
[error] ^
[error] somepathwithmyfile: value fromSource is not a member of org.apache.flink.api.scala.ExecutionEnvironment
[error] val text = env.fromSource(kafkaConsumer)
[error] ^
[error] two errors found
uj5u.com熱心網友回復：
對于第一個問題，洗掉new：
val kafkaConsumer = KafkaSource.builder[String]
...
對于第二個問題，fromSource需要三個引數：
/** Create a DataStream using a [[Source]]. */
@Experimental
def fromSource[T: TypeInformation](
source: Source[T, _ <: SourceSplit, _],
watermarkStrategy: WatermarkStrategy[T],
sourceName: String): DataStream[T] = {
val typeInfo = implicitly[TypeInformation[T]]
asScalaStream(javaEnv.fromSource(source, watermarkStrategy, sourceName, typeInfo))
}
另外，請注意 Flink 不（還）支持 scala 2.12.15。請參閱https://issues.apache.org/jira/browse/FLINK-20969。但是，如果排除 Flink 的內置 scala API 支持，Flink 1.15 可以與更新版本的 Scala（包括 Scala 3）一起使用。有關更多資訊，請參見https://flink.apache.org/2022/02/22/scala-free.html。
轉載請註明出處，本文鏈接：https://www.uj5u.com/net/512160.html