• 从RideCleansingSolution开始
  • 目标
    • 分割TaxiRide消息流
      • 一个流仅包含在纽约市开始和结束的行程（如前所述）
      • 另一条流包含其他行程（其他城市的行程）
  • 期待输出
    • 一个DataStream打印到stdout（纽约市内行程）
    • 一个DataStream打印到stderr(纽约市外行程)
      • 使用new PrintSinkFunction<>(true) 来创建一个sink到stderr
  • 注意
    • 这种拆分可以通过 FilterFunctions 或旁路输出来实现
    • 尝试并比较这两种方法

解决方案一：filter

  1. public class RideSplitFilterSolution {
  3.     private final SourceFunction<TaxiRide> source;
  4.     private final SinkFunction<TaxiRide> sink;
  5.     private final SinkFunction<TaxiRide> sidesink;
  7.     /** Creates a job using the source and sinks provided. */
  8.     public RideSplitFilterSolution(
  9.             SourceFunction<TaxiRide> source,
  10.             SinkFunction<TaxiRide> sink,
  11.             SinkFunction<TaxiRide> sidesink) {
  13.         this.source = source;
  14.         this.sink = sink;
  15.         this.sidesink = sidesink;
  16.     }
  18.     /**
  19.      * Main method.
  20.      *
  21.      * @throws Exception which occurs during job execution.
  22.      */
  23.     public static void main(String[] args) throws Exception {
  24.         RideSplitFilterSolution job =
  25.                 new RideSplitFilterSolution(
  26.                         new TaxiRideGenerator(),
  27.                         new PrintSinkFunction<>(),
  28.                         new PrintSinkFunction<>(true));
  30.         job.execute();
  31.     }
  1.  public JobExecutionResult execute() throws Exception {
  3.         // set up streaming execution environment
  4.         Configuration conf = new Configuration();
  5.         StreamExecutionEnvironment env =
  6.                 StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(conf);
  8.         // attach a stream of TaxiRides
  9.         DataStream<TaxiRide> rides = env.addSource(source);
  11.         // split the stream
  12.         rides.filter(ride -> inNYC(ride)).addSink(sink).name("inside NYC");
  13.         rides.filter(ride -> !inNYC(ride)).addSink(sidesink).name("outside NYC");
  15.         // run the pipeline and return the result
  16.         return env.execute("Split with filters");
  17.     }
  19.     /** Return true for rides that both start and end in NYC. */
  20.     private static boolean inNYC(TaxiRide taxiRide) {
  21.         return GeoUtils.isInNYC(taxiRide.startLon, taxiRide.startLat)
  22.                 && GeoUtils.isInNYC(taxiRide.endLon, taxiRide.endLat);
  23.     }

这是使用一组 FilterFunction 的解决方案的核心。我们已经分叉了行程，通过过滤器发送该流的一个副本，该过滤器保持在纽约市内开始和结束的行程，并通过执行相反操作的过滤器发送另一个副本。我们为接收器sink命名，以便在 Flink 仪表板中轻松识别它们。
image.png
我们将整个流的完整副本发送到两个过滤器。 因为在管道中的这一点上正在改变并行性，所以这很昂贵。 使用此解决方案，sink和filter构成chain。

解决方案二：基于旁路输出

  1. public class RideSplitSideOutputSolution {
  3.     private final SourceFunction<TaxiRide> source;
  4.     private final SinkFunction<TaxiRide> sink;
  5.     private final SinkFunction<TaxiRide> sidesink;
  7.     private static final OutputTag<TaxiRide> outsideNYC = new OutputTag<TaxiRide>("outsideNYC") {};
  9.     /** Creates a job using the source and sinks provided. */
  10.     public RideSplitSideOutputSolution(
  11.             SourceFunction<TaxiRide> source,
  12.             SinkFunction<TaxiRide> sink,
  13.             SinkFunction<TaxiRide> sidesink) {
  15.         this.source = source;
  16.         this.sink = sink;
  17.         this.sidesink = sidesink;
  18.     }
  20.     /**
  21.      * Main method.
  22.      *
  23.      * @throws Exception which occurs during job execution.
  24.      */
  25.     public static void main(String[] args) throws Exception {
  26.         RideSplitSideOutputSolution job =
  27.                 new RideSplitSideOutputSolution(
  28.                         new TaxiRideGenerator(),
  29.                         new PrintSinkFunction<>(),
  30.                         new PrintSinkFunction<>(true));
  32.         job.execute();
  33.     }
  35.     /**
  36.      * Creates and executes the long rides pipeline.
  37.      *
  38.      * @return {JobExecutionResult}
  39.      * @throws Exception which occurs during job execution.
  40.      */
  41.     public JobExecutionResult execute() throws Exception {
  42.         // set up streaming execution environment
  43.         Configuration conf = new Configuration();
  44.         StreamExecutionEnvironment env =
  45.                 StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(conf);
  47.         // attach a stream of TaxiRides
  48.         DataStream<TaxiRide> rides = env.addSource(source);
  50.         // split the stream
  51.         SingleOutputStreamOperator<TaxiRide> splitResult = rides.process(new StreamSplitter());
  53.         splitResult.addSink(sink).name("inside NYC");
  54.         splitResult.getSideOutput(outsideNYC).addSink(sidesink).name("outside NYC");
  56.         // run the pipeline and return the result
  57.         return env.execute("Split with side output");
  58.     }
  60.     public static class StreamSplitter extends ProcessFunction<TaxiRide, TaxiRide> {
  62.         @Override
  63.         public void processElement(
  64.                 TaxiRide taxiRide,
  65.                 Context ctx,
  66.                 Collector<TaxiRide> out)
  67.                 throws Exception {
  69.             if (GeoUtils.isInNYC(taxiRide.startLon, taxiRide.startLat)
  70.                     && GeoUtils.isInNYC(taxiRide.endLon, taxiRide.endLat)) {
  72.                 out.collect(taxiRide);
  73.             } else {
  74.                 ctx.output(outsideNYC, taxiRide);
  75.             }
  76.         }
  77.     }
  78. }

因为我们不需要keyed state或计时器，所以我们不需要使用 keyBy，我们将使用 ProcessFunction 而不是 KeyedProcessFunction。

使用side output是这个方案的核心，分割逻辑在StreamSplitter()中。如果这是一个n-way split，这个方案将比维持1组n个filter函数要优雅一些。
image.png
在process中执行了split逻辑，job graph相当简单。注意普通sink和side output sink都在同一个task（process函数）中。