title: Implementing a Streaming Cloud Dataflow transformation that calls an API for each element in the stream
body: <p><strong>Use Case</strong>:
  I have a Google Cloud Dataflow pipeline that processes a stream coming from PubSub. One of the transformations in the pipeline needs to read from a file that kas key value pairs. The size of the file is 130.5 MB. The file works as a lookup file where one column works as they key for the lookup. This is a pubsub to bigquery streaming pipeline.</p>
  <p><strong>My current Implementation:</strong>
  I have created an API that is hosted on App Engine. This API takes in a key as a parameter, searches the file and outputs the information based on that key.</p>
  <p>My dataflow pipeline has a transformation that calls this API for each and every element of the stream that I get from pubsub. This currently amounts to an average of 1,000,000 calls to the API from the Dataflow job each day. I am expecting this number to grow significantly.</p>
  <p><strong>Problem</strong>:
  This transformation has broken down multiple times. I consistently get network issues in the dataflow VMs. This causes the dataflow job to cease processing data and get stuck. This transformation has become a bottleneck for the functioning of the pipeline. As a safeguard, I have set a timeout of 5 seconds on each call to the API. I notice that the API timeouts about 5-6 times each hour.</p>
  <p>I believe that this is not the best way to implement this transformation. I am wondering if there is a better way to do this lookup from a dataflow job.</p>
  <p><strong>Follow up:</strong></p>
  <p>Trying to implement the use case as side input:</p>
  <p>The file that I am trying to read is GeoIP2-City.mmdb file. Based on MaxMind documentation, I am using the following code to create a reader object for the file. I want to pass in the DatabaseReader Object as a side input because I don’t want to create this object for each element that I get from the stream.</p>
  <pre><code>&gt; PcollectionView&lt;DatabaseReader&gt; reader = pipeline.apply(
  &gt;        ParDo.of(
  &gt;                new DoFn&lt;Long, DatabaseReader&gt;() {
  &gt;                    @ProcessElement
  &gt;                    public void process(
  &gt;                            @Element Long input, 
      OutputReceiver&lt;DatabaseReader&gt; o) throws IOException { //Following
  &gt; MaxMind Documentation to read file
  &gt;                            File database = new File(“GeoIP2-City.mmdb”);
  &gt;                            DatabaseReader reader = new DatabaseReader
  &gt;                                    .Builder(database)
  &gt;                                    .fileMode(com.maxmind.db.Reader.FileMode.MEMORY)
  &gt;                                    .withCache(new CHMCache())
  &gt;                                    .build();
  &gt;                            o.output(reader);
  &gt;                    }
  &gt;                })) .apply(View.asSingleton());
  </code></pre>
  <p>Here is the code where this side input is used:</p>
  <pre><code>&gt; PCollection&lt;TableRow&gt; newTableRow =   
  &gt; tableRowBeforeTransformation.apply(&quot;mmdb lookup&quot;, ParDo.of(new
  &gt; MmdbReaderFn(reader)).withSideInputs(reader));
  </code></pre>
  <p>I have created a MmdbReaderFn PTransformation class that applies the lookup logic using the side input.</p>
  <pre><code>&gt; DatabaseReader reader; PCollectionView&lt;DatabaseReader&gt; readerView;
  &gt; 
  &gt; public MmdbReaderFn(PCollectionView readerView){    this.readerView =
  &gt; readerView;
  &gt; 
  &gt; }
  &gt; 
  &gt; @ProcessElement public void processElement(ProcessContext context)
  &gt; throws IOException {    TableRow tableRow = context.element();   
  &gt; reader = context.sideInput(readerView);    String ip =
  &gt; String.valueOf(tableRow.get(&quot;ip&quot;));    try {
  &gt;        InetAddress ipAddress = InetAddress.getByName(ip);
  &gt; 
  &gt;        CityResponse response = reader.city(ipAddress);
  &gt;        if(response == null) {
  &gt;            log.info(&quot;Invalid IP address&quot;);
  &gt;        } else {
  &gt; 
  &gt;            if(response.getCountry() != null &amp;&amp; response.getCountry().getName() != null) {
  &gt;                tableRow.set(COUNTRY_NAME, response.getCountry().getName());      // 'Unite States'
  &gt;            } else {
  &gt;                tableRow.set(COUNTRY_NAME, null);
  &gt;            }
  &gt; 
  &gt;           
  &gt;            if(response.getCity() != null &amp;&amp; response.getCity().getName() != null) {
  &gt;                tableRow.set(CITY_NAME, response.getCity().getName());
  &gt;            } else {
  &gt;                tableRow.set(CITY_NAME, null);
  &gt;            }
  &gt; 
  &gt;                } catch (Exception e) {
  &gt;        log.error(&quot;Error process GeoIP2&quot;, e);
  &gt; 
  &gt;    }    context.output(tableRow); }
  </code></pre>
  <p>However, I am getting the following error when trying to create the PCollectionView for the reader.</p>
  <p>java.lang.IllegalStateException: Unable to return a default Coder for ParDo(Anonymous)8/ParMultiDo(Anonymous).output [PCollection]. Correct one of the following root causes:
  No Coder has been manually specified;  you may do so using .setCoder().
  Inferring a Coder from the CoderRegistry failed: Unable to provide a Coder for com.maxmind.geoip2.DatabaseReader.
  Building a Coder using a registered CoderProvider failed.
  See suppressed exceptions for detailed failures.
  Using the default output Coder from the producing PTransform failed: PTransform.getOutputCoder called.</p>