title: Performance improvement for Spark streaming job
body: <p>I have a spark job which keeps streaming data from a kafka topic which contains IP addresses. This data is huge around 1M/sec</p>
  <p>This data needs to be correlated with 5 of the Postgres Db tables which contains the subnets. So, the IP address coming from the input kafka topic, has to be validated against all these 5 tables containing subnets.</p>
  <p><strong>A little background on how <em>subnet</em> and <em>ip</em> are related</strong></p>
  <ul>
  <li>An Subnet can contain 'n' number of IP's</li>
  <li>Ex:  103.22.238.173/30 will have an IP range of 103.22.238.173 - 103.22.238.174</li>
  </ul>
  <p>So, We have below inputs in my Spark app.</p>
  <ol>
  <li>Data coming from kafka for around 1M/sec</li>
  <li>5 Database tables with approx. 5M records in each. These tables are cached in Spark for better performance.</li>
  </ol>
  <p><strong>HOW it is implemented currently</strong></p>
  <ol>
  <li>Calculate the long value of input ip and add another column to the dataframe i.e IPLong
  Ex:</li>
  <li>Take all 5 postgres tables(imported as Dataframe), pass the subnet column from the dataframe to an UDF which return the lowerLong and upperLong for the subnet and add these as two new columns to the table Dataframes.</li>
  </ol>
  <p>Now, i use the spark SQl, to find the if the incoming ip belongs to any of the postgres tables subnets like below:</p>
  <p><code>SELECT A.device,A.office,A.subnet, B.* FROM DbtableDF A RIGHT JOIN kafkaRawDF B ON A.upperLong &lt;= B.IPLong AND A.lowerLong &gt;= B.IPLong</code></p>
  <p>This entire process works fine in terms of correlation , but as the input data is coming with 1M/sec and each of the postgres tables have around 5M+ records, the services becomes extremely slow and then blocked.</p>
  <p>Note: Resources are not an issue, here is my executer config</p>
  <pre><code> executor:
      cores: 2
      coreLimit: &quot;2&quot;
      instances: 22
      memory: &quot;50g&quot;
  </code></pre>
  <p>I am new to apache spark, Can someone please suggest any better way to do this?</p>