title: How to design finding lowerBound & upperBound for spark read statement to partition the incoming data?
body: <p>To load and partition the incoming data in spark, I am using the following syntax.</p>
  <pre><code>val dataframe = spark.read.format(&quot;jdbc&quot;)
            .option(&quot;url&quot;, url)
            .option(&quot;driver&quot;, driver)
            .option(&quot;user&quot;, user)
            .option(&quot;password&quot;, password)
            .option(&quot;dbtable&quot;, query)
            .option(&quot;partitionColumn&quot;, partitionColumn)
            .option(&quot;lowerBound&quot;, lowerBound_value)
            .option(&quot;upperBound&quot;, upperBound_value)
            .option(&quot;numPartitions&quot;, numPartitions)
            .option(&quot;fetchsize&quot;, 15000)
            .load()
  </code></pre>
  <p>The parameters <code>partitionColumn</code>, <code>lowerBound</code>, <code>upperBound</code>, <code>numPartitions</code> are used to optimise the performance of the job.</p>
  <p>I have a table of 1000 records &amp; an integer column that has serial numbers from 1 to 1000.
  I am first running <code>min</code> and <code>max</code> on that column to assign <code>min</code> value to <code>lowerBound</code> and <code>max</code> value to <code>upperBound</code>. The numPartitions parameter is given as 3 so that the incoming data is split into 3 different partitions evenly (or close to being even).</p>
  <p>The above design works well when there is less data. But I have a scenario as below.</p>
  <p>I have a table of 203 billion records with no integer column that contain unique/serial integers. Then there is a date column that has data spread across 5 years namely 2016-2021.
  In order to move the data faster, I am moving a month's data of each year everytime.
  This is the query I am using:</p>
  <pre><code>val query = s&quot;(select * from table where date_column &gt;= '${YearMonth.of(year.toInt, month).atDay(1).toString} and date_time &lt;= '${YearMonth.of(year.toInt, month).atEndOfMonth().toString} 23:59:59.999') as datadf&quot;
  </code></pre>
  <p>So the above query becomes:
  <code>select * from table where date_column &gt;= '2016-01-01' and date_time &lt;= '2016-01-31 23:59:59.999''</code>
  and so on with first and last day of each month for every year.</p>
  <p>This is a rough description of how my loop is:</p>
  <pre><code>(2016 to 2021) { year =&gt;
     (1 to 12) { month =&gt;
             val query = s&quot;(select * from table where date_column &gt;= '${YearMonth.of(year.toInt, month).atDay(1).toString} and date_time &lt;= '${YearMonth.of(year.toInt, month).atEndOfMonth().toString} 23:59:59.999') as datadf&quot;
             val dataframe = spark.read.format(&quot;jdbc&quot;)
                .option(&quot;url&quot;, url)
                .option(&quot;driver&quot;, driver)
                .option(&quot;user&quot;, user)
                .option(&quot;password&quot;, password)
                .option(&quot;dbtable&quot;, query)
                .option(&quot;partitionColumn&quot;, partitionColumn)
                .option(&quot;lowerBound&quot;, lowerBound_value)
                .option(&quot;upperBound&quot;, upperBound_value)
                .option(&quot;numPartitions&quot;, numPartitions)
                .option(&quot;fetchsize&quot;, 15000)
                .load()
     }
  }
  </code></pre>
  <p>To find out bounds, I am using the same filters of month and year as below:</p>
  <pre><code>val bounds = spark.read.format(&quot;jdbc&quot;)
            .option(&quot;url&quot;, url)
            .option(&quot;driver&quot;, driver)
            .option(&quot;user&quot;, user)
            .option(&quot;password&quot;, password)
            .option(&quot;dbtable&quot;, &quot;(select min(partitionColumn) as mn, max(partitionColum) as from tablename where date_column &gt;= '${YearMonth.of(year.toInt, month).atDay(1).toString} and date_time &lt;= '${YearMonth.of(year.toInt, month).atEndOfMonth().toString} 23:59:59.999') as boundsDF&quot;)
            .load()
  
  val lowerBound_value = bounds.select(&quot;mn&quot;).head.getInt(0)
  val upperBound_value = bounds.select(&quot;mx&quot;).head.getInt(0)
  </code></pre>
  <p>The issue comes here with finding the lower and upper bounds of the filtered data.
  Because of the huge data size, the query that runs min &amp; max on the partitionColumn with the given filters is taking way more time than writing the actual dataframe into hdfs.</p>
  <p>I tried giving random values there but observed data skew in the partitions while the tasks are running.</p>
  <p>It is mandatory to give min and max of then partitionColumn as lower and upper bounds for better data distribution ?
  If not, is there any way to specify lower and upper bounds instead of running a min &amp; max query on data ?</p>
  <p>Any help is much appreciated.</p>