我开始用 Scala 语言研究 Apache Flink 的 CEP 库，当我尝试通过执行CEP.pattern(input,pattern)来创建 PatternStream 时，如 https://ci.apache.org/projects/flink/flink-docs-stable/dev/libs/cep.html 教程所示，IDE 说它"无法解析重载方法"，指的是pattern方法。根据我分别用于创建输入和模式的readTextFile和Pattern[String].begin('line').where(_.length == 10)的实现，该方法的参数或泛型类型应该没有任何问题。
这是我写的代码。我知道它不完整，但是自从这个问题出现以来，我无论如何都无法完成它。
package FlinkCEPClasses
import java.util.Properties
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.cep.CEP
import org.apache.flink.cep.scala.pattern.Pattern
import org.apache.flink.core.fs.FileSystem
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.datastream.DataStream
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer
class FlinkCEPPipeline {
var props : Properties = new Properties()
var env : StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
env.setParallelism(1)
var input : DataStream[String] = env.readTextFile("/home/luca/Desktop/lines")
var patt : Pattern[String,String] = Pattern.begin[String]("igual").where(_.length == 10)
// Problem appears at the following line. A red subscript appears at the pattern method,
// saying the following: "Cannot resolve overloaded method"
var CEPstream = CEP.pattern(input,patt)
input.writeAsText("/home/luca/Desktop/flinkcepout",FileSystem.WriteMode.OVERWRITE)
env.execute()
这是我的build.sbt文件内容：
name := "FlinkCEP"
version := "0.1"
scalaVersion := "2.12.10"
// https://mvnrepository.com/artifact/org.apache.flink/flink-cep-scala
libraryDependencies += "org.apache.flink" %% "flink-cep-scala" % "1.9.0"
// https://mvnrepository.com/artifact/org.apache.flink/flink-scala
libraryDependencies += "org.apache.flink" %% "flink-scala" % "1.9.0"
// https://mvnrepository.com/artifact/org.apache.flink/flink-streaming-scala
libraryDependencies += "org.apache.flink" %% "flink-streaming-scala" % "1.9.0"
// https://mvnrepository.com/artifact/org.apache.flink/flink-connector-kafka
libraryDependencies += "org.apache.flink" %% "flink-connector-kafka" % "1.9.0"
libraryDependencies += "log4j" % "log4j" % "1.2.17"
libraryDependencies += "ch.qos.logback" % "logback-classic" % "1.1.3" % Runtime
libraryDependencies += "org.slf4j" % "slf4j-simple" % "1.6.2" % Test```
我使用此代码的目的只是看到它运行一个简单的"where"条件，除此之外它不应该有任何更大的效用。我正在使用IntelliJ作为IDE。另外，我不确定 Scala 的 CEP 库是否已准备好使用。如果有人能对此有所了解，我将不胜感激。
试试这个：
import org.apache.flink.cep.scala.PatternStream
...
val CEPstream: PatternStream[String] = CEP.pattern[String](input, patt)
请参阅 github 以获取将 CEP 与 Scala 一起使用的简单示例。
好吧，在看了@DavidAnderson的 github 示例后，我终于解决了这个问题。由于我进行了许多更改，因此我无法确定我的解决方案是否适合您，但是我从import org.apache.flink.streaming.api.datastream.DataStream更改为import org.apache.flink.streaming.api.scala.{StreamExecutionEnvironment, DataStream, _}。注意不明确的导入，并确保导入真正的必需类。
我将列出我的所有导入和我的 build.sbt 文件，以便您可以完全访问我的配置。
进口
import java.util.Properties
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.cep.scala.pattern.Pattern
import org.apache.flink.core.fs.FileSystem
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.cep.scala.PatternStream
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer
import org.apache.flink.cep.scala.{CEP, PatternStream}
import org.apache.flink.streaming.api.scala.{StreamExecutionEnvironment, DataStream, _}
Build.sbt
name := "FlinkCEP"
version := "0.1"
scalaVersion := "2.12.10"
// https://mvnrepository.com/artifact/org.apache.flink/flink-cep-scala
//libraryDependencies += "org.apache.flink" %% "flink-cep-scala" % "1.9.0"
// https://mvnrepository.com/artifact/org.apache.flink/flink-cep
libraryDependencies += "org.apache.flink" %% "flink-cep" % "1.9.0"
// https://mvnrepository.com/artifact/org.apache.flink/flink-cep-scala
libraryDependencies += "org.apache.flink" %% "flink-cep-scala" % "1.9.0"
// https://mvnrepository.com/artifact/org.apache.flink/flink-runtime
libraryDependencies += "org.apache.flink" %% "flink-runtime" % "1.9.0" % Test
// https://mvnrepository.com/artifact/org.apache.flink/flink-scala
libraryDependencies += "org.apache.flink" %% "flink-scala" % "1.9.0"
// https://mvnrepository.com/artifact/org.apache.flink/flink-streaming-scala
libraryDependencies += "org.apache.flink" %% "flink-streaming-scala" % "1.9.0"
// https://mvnrepository.com/artifact/org.apache.flink/flink-connector-kafka
libraryDependencies += "org.apache.flink" %% "flink-connector-kafka" % "1.9.0"
libraryDependencies += "log4j" % "log4j" % "1.2.17"
libraryDependencies += "ch.qos.logback" % "logback-classic" % "1.1.3" % Runtime
libraryDependencies += "org.slf4j" % "slf4j-simple" % "1.6.2" % Test