title: Apache Flink - use values from a data stream to dynamically create a streaming data source
body: <p>I'm trying to build a sample application using Apache Flink that does the following:</p>
  
  <ol>
  <li>Reads a stream of stock symbols (e.g. 'CSCO', 'FB') from a Kafka queue.</li>
  <li>For each symbol performs a real-time lookup of current prices and streams the values for downstream processing.</li>
  </ol>
  
  <p><strong>* Update to original post *</strong></p>
  
  <p>I moved the map function into a separate class and do not get the run-time error message "<em>The implementation of the MapFunction is not serializable any more. The object probably contains or references non serializable fields</em>".</p>
  
  <p>The issue I'm facing now is that the Kafka topic "stockprices" I'm trying to write the prices to is not receiving them. I'm trying to trouble-shoot and will post any updates. </p>
  
  <pre><code>public class RetrieveStockPrices { 
      @SuppressWarnings("serial") 
      public static void main(String[] args) throws Exception { 
          final StreamExecutionEnvironment streamExecEnv = StreamExecutionEnvironment.getExecutionEnvironment();
          streamExecEnv.setStreamTimeCharacteristic(TimeCharacteristic.IngestionTime); 
  
          Properties properties = new Properties(); 
          properties.setProperty("bootstrap.servers", "localhost:9092"); 
          properties.setProperty("zookeeper.connect", "localhost:2181"); 
          properties.setProperty("group.id", "stocks"); 
  
          DataStream&lt;String&gt; streamOfStockSymbols = streamExecEnv.addSource(new FlinkKafkaConsumer08&lt;String&gt;("stocksymbol", new SimpleStringSchema(), properties)); 
  
          DataStream&lt;String&gt; stockPrice = 
              streamOfStockSymbols 
              //get unique keys 
              .keyBy(new KeySelector&lt;String, String&gt;() { 
                  @Override 
                  public String getKey(String trend) throws Exception {
                      return trend; 
                  }
                  }) 
              //collect events over a window 
              .window(TumblingEventTimeWindows.of(Time.seconds(60))) 
              //return the last event from the window...all elements are the same "Symbol" 
              .apply(new WindowFunction&lt;String, String, String, TimeWindow&gt;() {
                  @Override 
                  public void apply(String key, TimeWindow window, Iterable&lt;String&gt; input, Collector&lt;String&gt; out) throws Exception { 
                      out.collect(input.iterator().next().toString()); 
                  }
              })
              .map(new StockSymbolToPriceMapFunction());
  
          streamExecEnv.execute("Retrieve Stock Prices"); 
      }
  }
  
  public class StockSymbolToPriceMapFunction extends RichMapFunction&lt;String, String&gt; {
      @Override
      public String map(String stockSymbol) throws Exception {
          final StreamExecutionEnvironment streamExecEnv = StreamExecutionEnvironment.getExecutionEnvironment();
          streamExecEnv.setStreamTimeCharacteristic(TimeCharacteristic.IngestionTime);
          System.out.println("StockSymbolToPriceMapFunction: stockSymbol: " + stockSymbol);
  
          DataStream&lt;String&gt; stockPrices = streamExecEnv.addSource(new LookupStockPrice(stockSymbol));
          stockPrices.keyBy(new CustomKeySelector()).addSink(new FlinkKafkaProducer08&lt;String&gt;("localhost:9092", "stockprices", new SimpleStringSchema()));
  
          return "100000";
      }
  
      private static class CustomKeySelector implements KeySelector&lt;String, String&gt; {
          @Override
          public String getKey(String arg0) throws Exception {
              return arg0.trim();
          }
      }
  }
  
  
  public class LookupStockPrice extends RichSourceFunction&lt;String&gt; { 
      public String stockSymbol = null; 
      public boolean isRunning = true; 
  
      public LookupStockPrice(String inSymbol) { 
              stockSymbol = inSymbol; 
      } 
  
      @Override 
      public void open(Configuration parameters) throws Exception { 
              isRunning = true; 
      } 
  
  
      @Override 
      public void cancel() { 
              isRunning = false; 
      } 
  
      @Override 
      public void run(SourceFunction.SourceContext&lt;String&gt; ctx) 
                      throws Exception { 
              String stockPrice = "0";
              while (isRunning) { 
                  //TODO: query Google Finance API 
                  stockPrice = Integer.toString((new Random()).nextInt(100)+1);
                  ctx.collect(stockPrice);
                  Thread.sleep(10000);
              } 
      } 
  }
  </code></pre>