无法在flatMap()函数内创建DataStream？执行无输出求解

核心原因 #

Flink的作业执行逻辑是先构建完整的数据流拓扑，再通过env.execute()触发整个拓扑的运行，你在flatMap算子内部创建的newStream属于无效的孤立流，不会被执行：

• flatMap方法是算子运行时的处理逻辑，此时作业拓扑已经构建完成并等待执行，在这个阶段动态创建的DataStream不会被加入到当前作业的拓扑中。
• newStream.print()只是定义了一个带Sink的数据流，但没有任何机制触发这条流的执行——整个作业仅会执行env.execute()启动的主拓扑（也就是从初始data流到最终print的链路），孤立的流不会被调度运行。

所以你看到的-----之间没有输出，完全符合这个逻辑：newStream.print()根本没被执行，只有主拓扑的print输出了8> test message。

正确的实现方式 #

如果需要在flatMap里处理并输出数据，不需要创建新的DataStream，直接用以下两种方式：

1. 直接用System.out.println()打印：

@Override
public void flatMap(String s, Collector<String> collector) {
    System.out.println("-----");
    System.out.println(s);
    System.out.println("-----");
    collector.collect(s);
}

2. 通过Collector将数据收集到主数据流中，由主数据流的print()输出：

@Override
public void flatMap(String s, Collector<String> collector) {
    System.out.println("-----");
    collector.collect(s); // 数据进入主流，由外部的print输出
    System.out.println("-----");
}

内容的提问来源于stack exchange，提问作者Noma