我试图写一个流应用程序，既读取和写入Kafka。我目前有这个，但我必须字符串我的元组类。

object StreamingJob {
  def main(args: Array[String]) {
    // set up the streaming execution environment
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    val properties = new Properties()
    properties.setProperty("bootstrap.servers", "localhost:9092")
    properties.setProperty("zookeeper.connect", "localhost:2181")
    properties.setProperty("group.id", "test")

    val consumer = env.addSource(new FlinkKafkaConsumer08[String]("topic", new SimpleStringSchema(), properties))

    val counts = consumer.flatMap { _.toLowerCase.split("\\W+") filter { _.nonEmpty } }
      .map { (_, 1) }
      .keyBy(0)
      .timeWindow(Time.seconds(5))
      .sum(1)

    val producer = new FlinkKafkaProducer08[String](
      "localhost:9092",
      "my-topic",
      new SimpleStringSchema())

    counts.map(_.toString()).addSink(producer)

    env.execute("Window Stream WordCount")
    env.execute("Flink Streaming Scala API Skeleton")
  }
}

我能得到的最接近这个工作的是下面的代码，但是flinkkafkaproducer08拒绝接受类型参数作为构造函数的一部分。

val producer = new FlinkKafkaProducer08[(String, Int)](
  "localhost:9092",
  "my-topic",
  new TypeSerializerOutputFormat[(String, Int)])

counts.addSink(producer)

我想知道有没有办法把元组直接写到我的KafkaFlume里。