51zero / 51zero/eel-sdk

JdbcSource to partition queries for potential performance improvements

未关闭
#236 14 条评论 0 个 reaction 已指派 1 人 已被 @hannesmiller 认领 在 GitHub 查看
主要语言
Scala
星标
147
派生
32
PR 合并指标
30 天内没有已合并 PR

描述

- 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();
}
}
```

贡献指南

这个仓库没有索引到贡献指南

调研方向

The JdbcSource class likely handles JDBC connections and query execution. Look for methods that build queries and manage result sets. The proposal involves modifying the query generation to add partition predicates and potentially handling multiple result sets. Start by examining the existing JDBC integration and Spark partitioning parameters. Testing would require setting up a database and verifying the partitioned queries work correctly across different database systems like Oracle and SQL Server.

由索引模型根据 Issue 内容生成。

评估

技术栈
java, mysql, postgresql, scala, spark, sql
领域
backend, data-engineering, databases, performance
Issue 类型
功能
难度
4/5
预计耗时
3-5 天
活跃度
停滞
描述清晰度
基本清楚
新手友好度
30/100

把新 issue 发到你的邮箱

精选适合新手参与的 GitHub issue 摘要。