使用datastream api，您可以简单地执行以下操作：

streamFromKafka = env.addSource(new FlinkKafkaConsumer<>(inputTopic, ...));

streamFromKafka
  ... // one set of operations
  .addSink(new FlinkKafkaProducer<>(outputTopic1, ...));

streamFromKafka
  ... // another set of operations
  .addSink(new FlinkKafkaProducer<>(outputTopic2, ...));

为了使用表api，您必须对此进行一些调整，但原理是相同的。