我想要的是：阅读kafka主题，解析它的消息，例如，使用processfunction根据特定kafka消息字段的值创建新的executionenvironment（新的子图）。喜欢： IF (field.equals('readMongo')) THEN ExecutionEnvironment env = ..getExecutionEnvironment() ;......; env.execute(); 某种：持续监视kafka主题（流）-->执行特定的flink（数据集或无界流）任务。Kafka主题将每天定期但非常低频的信息：每天一次和两次。谢谢。