51zero / 51zero/eel-sdk

JdbcSource to partition queries for potential performance improvements

Open
#236 14 comments 0 reactions 1 assignee Claimed by @hannesmiller View on GitHub
Dominant language
Scala
Stars
147
Forks
32
PR merge metrics
No merged PRs in 30d

Description

- Spark can partition data on a **JDBC** data frame by by specifying the following binding parameters which are all **longs**: **lowerBound**, **upperBound**, **numPartitions** and and **partition key** column
- To take advantage of this in **JdbcSource**, the data has to be divided into multiple partitions (in multiple Threads). In turn these binding parameters are used to alter the original query for each partition, e.g. add a predicate to restrict query on each partition - for **Oracle** the partition key could be **rownum** which is available on every table, e.g. for a population consisting of 100 rows with a specification of: **lowerBound=1**, **upperBound=100** and **numPartitions=5** and **query=select col1, col2 from table where blah** would result in the following partitions queries on each partition:

## Part 1:
```sql
select * (select col1, col2 from table where blah)
where rownum between 1 and 20
```
## Part 2:
```sql
select * (select col1, col2 from table where blah)
where rownum between 21 and 40
```
## Part 3:
```sql
select * (select col1, col2 from table where blah)
where rownum between 41 and 60
```
## Part 4:
```sql
select * (select col1, col2 from table where blah)
where rownum between 61 and 80
```
## Part 5:
```sql
select * (select col1, col2 from table where blah)
where rownum >= 81
```
- For **partition 5** just return the remainder of rows.

- Note it may not be necessary to create N connections for each partition - simply return N **JDBC** result sets - one for each partition - investigating this...

# Proposal
- **withPartition(lowerBound, upperBound, partitionColumn)**
```scala
.withPartition(1, 100, rownum)
```
- **rownum** is the partition key which is internal to **Oracle** however you can't use this across the board with a function like **withPartition(1,100)**, e.g. **SQLServer** doesn't have **rownum**, however it can be achieved using a the windowing function **ROW_NUMBER()**:
```sql
where RowNumber = ROW_NUMBER() OVER (ORDER BY CustomerID ASC)
```
- where **CustomerID** is the primary key and **ROW_NUMBER()** is the **SQLServer** windowing function.
- One idea is to provide a upper bound function like so:
```scala
withPartition(lowerBound:Long, upperBoundFn(query:String) => Long, partitionColumn:String)
```
- the **upperBoundFn()** could do anything you like such as execute a separate query or just supply the count.

# Example of returning N JDBC result sets

- http://stackoverflow.com/questions/9696572/queries-returning-multiple-result-sets

```java

public static void executeProcedure(Connection con) {
try {
CallableStatement stmt = con.prepareCall(...);
..... //Set call parameters, if you have IN,OUT, or IN/OUT parameters

boolean results = stmt.execute();
int rsCount = 0;

//Loop through the available result sets.
while (results) {
ResultSet rs = stmt.getResultSet();
//Retrieve data from the result set.
while (rs.next()) {
....// using rs.getxxx() method to retieve data
}
rs.close();

//Check for next result set
results = stmt.getMoreResults();
}
stmt.close();
}
catch (Exception e) {
e.printStackTrace();
}
}
```

Contributor guide

No contributing guide indexed for this repository

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.