title: How do I join two streams in apache flink?
body: <p>I am getting started with flink and having a look at <a href="https://github.com/apache/flink-training/tree/master/rides-and-fares" rel="nofollow noreferrer">one of the official tutorials</a>. </p>
  
  <p>To my understanding the goal of this exercise is to join the two streams on the time attribute.</p>
  
  <p><strong>Task:</strong></p>
  
  <blockquote>
    <p>The result of this exercise is a data stream of Tuple2 records, one for each distinct rideId. You should ignore the
    END events, and only join the event for the START of each ride with
    its corresponding fare data.</p>
    
    <p>The resulting stream should be printed to standard out.</p>
  </blockquote>
  
  <p><strong>Question:</strong> How is the EnrichmentFunction able to join the two streams aka. how does it know which fair to join with which ride? I expected it to buffer multiple fairs/rides until for an incoming fair/ride there is a matching partner.</p>
  
  <p>In my understanding it just saves every ride/fair it sees and combines it with the next best ride/fair. Why is this a proper join? </p>
  
  <p><strong>Provided Solution:</strong></p>
  
  <pre><code>/*
   * Copyright 2017 data Artisans GmbH
   *
   * Licensed under the Apache License, Version 2.0 (the "License");
   * you may not use this file except in compliance with the License.
   * You may obtain a copy of the License at
   *
   *  http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  
  package com.dataartisans.flinktraining.solutions.datastream_java.state;
  
  import com.dataartisans.flinktraining.exercises.datastream_java.datatypes.TaxiFare;
  import com.dataartisans.flinktraining.exercises.datastream_java.datatypes.TaxiRide;
  import com.dataartisans.flinktraining.exercises.datastream_java.sources.TaxiFareSource;
  import com.dataartisans.flinktraining.exercises.datastream_java.sources.TaxiRideSource;
  import com.dataartisans.flinktraining.exercises.datastream_java.utils.ExerciseBase;
  import org.apache.flink.api.common.state.ValueState;
  import org.apache.flink.api.common.state.ValueStateDescriptor;
  import org.apache.flink.api.java.tuple.Tuple2;
  import org.apache.flink.api.java.utils.ParameterTool;
  import org.apache.flink.configuration.Configuration;
  import org.apache.flink.streaming.api.TimeCharacteristic;
  import org.apache.flink.streaming.api.datastream.DataStream;
  import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  import org.apache.flink.streaming.api.functions.co.RichCoFlatMapFunction;
  import org.apache.flink.util.Collector;
  
  /**
   * Java reference implementation for the "Stateful Enrichment" exercise of the Flink training
   * (http://training.data-artisans.com).
   *
   * The goal for this exercise is to enrich TaxiRides with fare information.
   *
   * Parameters:
   * -rides path-to-input-file
   * -fares path-to-input-file
   *
   */
  public class RidesAndFaresSolution extends ExerciseBase {
      public static void main(String[] args) throws Exception {
  
          ParameterTool params = ParameterTool.fromArgs(args);
          final String ridesFile = params.get("rides", pathToRideData);
          final String faresFile = params.get("fares", pathToFareData);
  
          final int delay = 60;                   // at most 60 seconds of delay
          final int servingSpeedFactor = 1800;    // 30 minutes worth of events are served every second
  
          // set up streaming execution environment
          StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
          env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
          env.setParallelism(ExerciseBase.parallelism);
  
          DataStream&lt;TaxiRide&gt; rides = env
                  .addSource(rideSourceOrTest(new TaxiRideSource(ridesFile, delay, servingSpeedFactor)))
                  .filter((TaxiRide ride) -&gt; ride.isStart)
                  .keyBy("rideId");
  
          DataStream&lt;TaxiFare&gt; fares = env
                  .addSource(fareSourceOrTest(new TaxiFareSource(faresFile, delay, servingSpeedFactor)))
                  .keyBy("rideId");
  
          DataStream&lt;Tuple2&lt;TaxiRide, TaxiFare&gt;&gt; enrichedRides = rides
                  .connect(fares)
                  .flatMap(new EnrichmentFunction());
  
          printOrTest(enrichedRides);
  
          env.execute("Join Rides with Fares (java RichCoFlatMap)");
      }
  
      public static class EnrichmentFunction extends RichCoFlatMapFunction&lt;TaxiRide, TaxiFare, Tuple2&lt;TaxiRide, TaxiFare&gt;&gt; {
          // keyed, managed state
          private ValueState&lt;TaxiRide&gt; rideState;
          private ValueState&lt;TaxiFare&gt; fareState;
  
          @Override
          public void open(Configuration config) {
              rideState = getRuntimeContext().getState(new ValueStateDescriptor&lt;&gt;("saved ride", TaxiRide.class));
              fareState = getRuntimeContext().getState(new ValueStateDescriptor&lt;&gt;("saved fare", TaxiFare.class));
          }
  
          @Override
          public void flatMap1(TaxiRide ride, Collector&lt;Tuple2&lt;TaxiRide, TaxiFare&gt;&gt; out) throws Exception {
              TaxiFare fare = fareState.value();
              if (fare != null) {
                  fareState.clear();
                  out.collect(new Tuple2(ride, fare));
              } else {
                  rideState.update(ride);
              }
          }
  
          @Override
          public void flatMap2(TaxiFare fare, Collector&lt;Tuple2&lt;TaxiRide, TaxiFare&gt;&gt; out) throws Exception {
              TaxiRide ride = rideState.value();
              if (ride != null) {
                  rideState.clear();
                  out.collect(new Tuple2(ride, fare));
              } else {
                  fareState.update(fare);
              }
          }
      }
  }
  </code></pre>