我只是在flink中尝试使用scala类型的类。我定义了以下类型类接口：

trait LikeEvent[T] {
    def timestamp(payload: T): Int
}

现在，我想考虑 DataSet 的 LikeEvent[_] 这样地：

// existing classes that need to be adapted/normalized (without touching them)
case class Log(ts: Int, severity: Int, message: String)
case class Metric(ts: Int, name: String, value: Double)

// create instances for the raw events
object EventInstance {

    implicit val logEvent = new LikeEvent[Log] {
        def timestamp(log: Log): Int = log.ts
    }

    implicit val metricEvent = new LikeEvent[Metric] {
        def timestamp(metric: Metric): Int = metric.ts
    }
}

// add ops to the raw event classes (regular class)
object EventSyntax {

    implicit class Event[T: LikeEvent](val payload: T) {
        val le = implicitly[LikeEvent[T]]
        def timestamp: Int = le.timestamp(payload)
    }
}

以下应用程序运行正常：

// set up the execution environment
val env = ExecutionEnvironment.getExecutionEnvironment

// underlying (raw) events
val events: DataSet[Event[_]] = env.fromElements(
  Metric(1586736000, "cpu_usage", 0.2),
  Log(1586736005, 1, "invalid login"),
  Log(1586736010, 1, "invalid login"),
  Log(1586736015, 1, "invalid login"),
  Log(1586736030, 2, "valid login"),
  Metric(1586736060, "cpu_usage", 0.8),
  Log(1586736120, 0, "end of world"),
)

// count events per hour
val eventsPerHour = events
  .map(new GetMinuteEventTuple())
  .groupBy(0).reduceGroup { g =>
    val gl = g.toList
    val (hour, count) = (gl.head._1, gl.size)
    (hour, count)
  }

eventsPerHour.print()

打印预期输出

(0,5)
(1,1)
(2,1)

但是，如果我这样修改syntax对象：

// couldn't make it work with Flink!
// add ops to the raw event classes (case class)
object EventSyntax2 {

  case class Event[T: LikeEvent](payload: T) {
    val le = implicitly[LikeEvent[T]]
    def timestamp: Int = le.timestamp(payload)
  }

  implicit def fromPayload[T: LikeEvent](payload: T): Event[T] = Event(payload)  
}

我得到以下错误：

type mismatch;
found   : org.apache.flink.api.scala.DataSet[Product with Serializable]
required: org.apache.flink.api.scala.DataSet[com.salvalcantara.fp.EventSyntax2.Event[_]]

因此，在这个信息的指引下，我做了以下改变：

val events: DataSet[Event[_]] = env.fromElements[Event[_]](...)

之后，错误变为：

could not find implicit value for evidence parameter of type org.apache.flink.api.common.typeinfo.TypeInformation[com.salvalcantara.fp.EventSyntax2.Event[_]]

我不明白为什么 EventSyntax2 导致这些错误 EventSyntax 编译和运行良好。为什么要在中使用case类 Package 器 EventSyntax2 比在中使用常规类更有问题 EventSyntax ?
不管怎样，我的问题有两个：
我怎样才能解决我的问题 EventSyntax2 ?
实现我的目标最简单的方法是什么？在这里，我只是为了学习而尝试使用类型类模式，但是一种更面向对象的方法（基于子类型）在我看来更简单。像这样：

// Define trait
trait Event {
    def timestamp: Int
    def payload: Product with Serializable // Any case class
}

// Metric adapter (similar for Log)
object MetricAdapter {

    implicit class MetricEvent(val payload: Metric) extends Event {
        def timestamp: Int = payload.ts
    }
}

然后简单地使用 val events: DataSet[Event] = env.fromElements(...) 总的来说。
注意，实现某个typeclass的类列表提出了一个类似的问题，但它考虑了一个简单的scala List 而不是燧石 DataSet （或 DataStream ). 我的问题的焦点是使用flink中的类型类模式来考虑异构流/数据集，以及它是否真的有意义，或者在这种情况下应该明确地支持一个常规特性并从中继承，如上所述。
顺便说一句，您可以在这里找到代码：https://github.com/salvalcantara/flink-events-and-polymorphism.