You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@spark.apache.org by Mike Artz <mi...@gmail.com> on 2020/05/25 05:20:25 UTC
Re: Parallelising JDBC reads in spark
Does anything different happened when you set the isolationLevel to do
Dirty Reads i.e. "READ_UNCOMMITTED"
On Sun, May 24, 2020 at 7:50 PM Manjunath Shetty H <ma...@live.com>
wrote:
> Hi,
>
> We are writing a ETL pipeline using Spark, that fetch the data from SQL
> server in batch mode (every 15mins). Problem we are facing when we try to
> parallelising single table reads into multiple tasks without missing any
> data.
>
> We have tried this,
>
>
> - Use `ROW_NUMBER` window function in the SQL query
> - Then do
> -
>
> DataFrame df =
> hiveContext
> .read()
> .jdbc(
> *<url>*,
> query,
> "row_num",
> 1,
> <upper_limit>,
> noOfPartitions,
> jdbcOptions);
>
>
>
> The problem with this approach is if our tables get updated in between in SQL Server while tasks are still running then the `ROW_NUMBER` will change and we may miss some records.
>
>
> Any approach to how to fix this issue ? . Any pointers will be helpful
>
>
> *Note*: I am on spark 1.6
>
>
> Thanks
>
> Manjiunath Shetty
>
>
Re: Parallelising JDBC reads in spark
Posted by Manjunath Shetty H <ma...@live.com>.
Thanks Dhaval for the suggestion, but in the case i mentioned in previous mail still data can be missed as the row number will change.
-
Manjunath
________________________________
From: Dhaval Patel <ma...@gmail.com>
Sent: Monday, May 25, 2020 3:01 PM
To: Manjunath Shetty H <ma...@live.com>
Subject: Re: Parallelising JDBC reads in spark
If possible, set the watermark before reading data. Read the max of watermark column before reading actual data and add that in query to read actual data, like watermark <= current_watermark
It may query db twice, however it will make sure you are not missing any records
Regards
Dhaval
On Mon, May 25, 2020 at 3:38 AM Manjunath Shetty H <ma...@live.com>> wrote:
Thanks Georg for the suggestion, but at this point changing the design is not really the option.
Any other pointer would be helpful.
Thanks
Manjunath
________________________________
From: Georg Heiler <ge...@gmail.com>>
Sent: Monday, May 25, 2020 11:52 AM
To: Manjunath Shetty H <ma...@live.com>>
Cc: Mike Artz <mi...@gmail.com>>; user <us...@spark.apache.org>>
Subject: Re: Parallelising JDBC reads in spark
Well you seem to have performance and consistency problems. Using a CDC tool fitting for your database you might be able to fix both.
However, streaming the change events of the database log might be a bit more complicated. Tools like https://debezium.io/ could be useful - depending on your source database.
Best,
Georg
Am Mo., 25. Mai 2020 um 08:16 Uhr schrieb Manjunath Shetty H <ma...@live.com>>:
Hi Georg,
Thanks for the response, can please elaborate what do mean by change data capture ?
Thanks
Manjunath
________________________________
From: Georg Heiler <ge...@gmail.com>>
Sent: Monday, May 25, 2020 11:14 AM
To: Manjunath Shetty H <ma...@live.com>>
Cc: Mike Artz <mi...@gmail.com>>; user <us...@spark.apache.org>>
Subject: Re: Parallelising JDBC reads in spark
Why don't you apply proper change data capture?
This will be more complex though.
Am Mo., 25. Mai 2020 um 07:38 Uhr schrieb Manjunath Shetty H <ma...@live.com>>:
Hi Mike,
Thanks for the response.
Even with that flag set data miss can happen right ?. As the fetch is based on the last watermark (maximum timestamp of the row that last batch job fetched ), Take a scenario like this with table
a : 1
b : 2
c : 3
d : 4
f : 6
g : 7
h : 8
e : 5
* a,b,c,d,e get picked by 1 task
* by the time second task starts, e has been updated, so the row order changes
* As f moves up, it will completely get missed in the fetch
Thanks
Manjunath
________________________________
From: Mike Artz <mi...@gmail.com>>
Sent: Monday, May 25, 2020 10:50 AM
To: Manjunath Shetty H <ma...@live.com>>
Cc: user <us...@spark.apache.org>>
Subject: Re: Parallelising JDBC reads in spark
Does anything different happened when you set the isolationLevel to do Dirty Reads i.e. "READ_UNCOMMITTED"
On Sun, May 24, 2020 at 7:50 PM Manjunath Shetty H <ma...@live.com>> wrote:
Hi,
We are writing a ETL pipeline using Spark, that fetch the data from SQL server in batch mode (every 15mins). Problem we are facing when we try to parallelising single table reads into multiple tasks without missing any data.
We have tried this,
* Use `ROW_NUMBER` window function in the SQL query
* Then do
*
DataFrame df =
hiveContext
.read()
.jdbc(
<url>,
query,
"row_num",
1,
<upper_limit>,
noOfPartitions,
jdbcOptions);
The problem with this approach is if our tables get updated in between in SQL Server while tasks are still running then the `ROW_NUMBER` will change and we may miss some records.
Any approach to how to fix this issue ? . Any pointers will be helpful
Note: I am on spark 1.6
Thanks
Manjiunath Shetty
Re: Parallelising JDBC reads in spark
Posted by Manjunath Shetty H <ma...@live.com>.
Thanks Georg for the suggestion, but at this point changing the design is not really the option.
Any other pointer would be helpful.
Thanks
Manjunath
________________________________
From: Georg Heiler <ge...@gmail.com>
Sent: Monday, May 25, 2020 11:52 AM
To: Manjunath Shetty H <ma...@live.com>
Cc: Mike Artz <mi...@gmail.com>; user <us...@spark.apache.org>
Subject: Re: Parallelising JDBC reads in spark
Well you seem to have performance and consistency problems. Using a CDC tool fitting for your database you might be able to fix both.
However, streaming the change events of the database log might be a bit more complicated. Tools like https://debezium.io/ could be useful - depending on your source database.
Best,
Georg
Am Mo., 25. Mai 2020 um 08:16 Uhr schrieb Manjunath Shetty H <ma...@live.com>>:
Hi Georg,
Thanks for the response, can please elaborate what do mean by change data capture ?
Thanks
Manjunath
________________________________
From: Georg Heiler <ge...@gmail.com>>
Sent: Monday, May 25, 2020 11:14 AM
To: Manjunath Shetty H <ma...@live.com>>
Cc: Mike Artz <mi...@gmail.com>>; user <us...@spark.apache.org>>
Subject: Re: Parallelising JDBC reads in spark
Why don't you apply proper change data capture?
This will be more complex though.
Am Mo., 25. Mai 2020 um 07:38 Uhr schrieb Manjunath Shetty H <ma...@live.com>>:
Hi Mike,
Thanks for the response.
Even with that flag set data miss can happen right ?. As the fetch is based on the last watermark (maximum timestamp of the row that last batch job fetched ), Take a scenario like this with table
a : 1
b : 2
c : 3
d : 4
f : 6
g : 7
h : 8
e : 5
* a,b,c,d,e get picked by 1 task
* by the time second task starts, e has been updated, so the row order changes
* As f moves up, it will completely get missed in the fetch
Thanks
Manjunath
________________________________
From: Mike Artz <mi...@gmail.com>>
Sent: Monday, May 25, 2020 10:50 AM
To: Manjunath Shetty H <ma...@live.com>>
Cc: user <us...@spark.apache.org>>
Subject: Re: Parallelising JDBC reads in spark
Does anything different happened when you set the isolationLevel to do Dirty Reads i.e. "READ_UNCOMMITTED"
On Sun, May 24, 2020 at 7:50 PM Manjunath Shetty H <ma...@live.com>> wrote:
Hi,
We are writing a ETL pipeline using Spark, that fetch the data from SQL server in batch mode (every 15mins). Problem we are facing when we try to parallelising single table reads into multiple tasks without missing any data.
We have tried this,
* Use `ROW_NUMBER` window function in the SQL query
* Then do
*
DataFrame df =
hiveContext
.read()
.jdbc(
<url>,
query,
"row_num",
1,
<upper_limit>,
noOfPartitions,
jdbcOptions);
The problem with this approach is if our tables get updated in between in SQL Server while tasks are still running then the `ROW_NUMBER` will change and we may miss some records.
Any approach to how to fix this issue ? . Any pointers will be helpful
Note: I am on spark 1.6
Thanks
Manjiunath Shetty
Re: Parallelising JDBC reads in spark
Posted by Georg Heiler <ge...@gmail.com>.
Well you seem to have performance and consistency problems. Using a CDC
tool fitting for your database you might be able to fix both.
However, streaming the change events of the database log might be a bit
more complicated. Tools like https://debezium.io/ could be useful -
depending on your source database.
Best,
Georg
Am Mo., 25. Mai 2020 um 08:16 Uhr schrieb Manjunath Shetty H <
manjunathshetty@live.com>:
> Hi Georg,
>
> Thanks for the response, can please elaborate what do mean by change data
> capture ?
>
> Thanks
> Manjunath
> ------------------------------
> *From:* Georg Heiler <ge...@gmail.com>
> *Sent:* Monday, May 25, 2020 11:14 AM
> *To:* Manjunath Shetty H <ma...@live.com>
> *Cc:* Mike Artz <mi...@gmail.com>; user <us...@spark.apache.org>
> *Subject:* Re: Parallelising JDBC reads in spark
>
> Why don't you apply proper change data capture?
> This will be more complex though.
>
> Am Mo., 25. Mai 2020 um 07:38 Uhr schrieb Manjunath Shetty H <
> manjunathshetty@live.com>:
>
> Hi Mike,
>
> Thanks for the response.
>
> Even with that flag set data miss can happen right ?. As the fetch is
> based on the last watermark (maximum timestamp of the row that last batch
> job fetched ), Take a scenario like this with table
>
> a : 1
> b : 2
> c : 3
> d : 4
> *f : 6*
> g : 7
> h : 8
> e : 5
>
>
> - a,b,c,d,e get picked by 1 task
> - by the time second task starts, e has been updated, so the row order
> changes
> - As f moves up, it will completely get missed in the fetch
>
>
> Thanks
> Manjunath
>
> ------------------------------
> *From:* Mike Artz <mi...@gmail.com>
> *Sent:* Monday, May 25, 2020 10:50 AM
> *To:* Manjunath Shetty H <ma...@live.com>
> *Cc:* user <us...@spark.apache.org>
> *Subject:* Re: Parallelising JDBC reads in spark
>
> Does anything different happened when you set the isolationLevel to do
> Dirty Reads i.e. "READ_UNCOMMITTED"
>
> On Sun, May 24, 2020 at 7:50 PM Manjunath Shetty H <
> manjunathshetty@live.com> wrote:
>
> Hi,
>
> We are writing a ETL pipeline using Spark, that fetch the data from SQL
> server in batch mode (every 15mins). Problem we are facing when we try to
> parallelising single table reads into multiple tasks without missing any
> data.
>
> We have tried this,
>
>
> - Use `ROW_NUMBER` window function in the SQL query
> - Then do
> -
>
> DataFrame df =
> hiveContext
> .read()
> .jdbc(
> *<url>*,
> query,
> "row_num",
> 1,
> <upper_limit>,
> noOfPartitions,
> jdbcOptions);
>
>
>
> The problem with this approach is if our tables get updated in between in SQL Server while tasks are still running then the `ROW_NUMBER` will change and we may miss some records.
>
>
> Any approach to how to fix this issue ? . Any pointers will be helpful
>
>
> *Note*: I am on spark 1.6
>
>
> Thanks
>
> Manjiunath Shetty
>
>
Re: Parallelising JDBC reads in spark
Posted by Manjunath Shetty H <ma...@live.com>.
Hi Georg,
Thanks for the response, can please elaborate what do mean by change data capture ?
Thanks
Manjunath
________________________________
From: Georg Heiler <ge...@gmail.com>
Sent: Monday, May 25, 2020 11:14 AM
To: Manjunath Shetty H <ma...@live.com>
Cc: Mike Artz <mi...@gmail.com>; user <us...@spark.apache.org>
Subject: Re: Parallelising JDBC reads in spark
Why don't you apply proper change data capture?
This will be more complex though.
Am Mo., 25. Mai 2020 um 07:38 Uhr schrieb Manjunath Shetty H <ma...@live.com>>:
Hi Mike,
Thanks for the response.
Even with that flag set data miss can happen right ?. As the fetch is based on the last watermark (maximum timestamp of the row that last batch job fetched ), Take a scenario like this with table
a : 1
b : 2
c : 3
d : 4
f : 6
g : 7
h : 8
e : 5
* a,b,c,d,e get picked by 1 task
* by the time second task starts, e has been updated, so the row order changes
* As f moves up, it will completely get missed in the fetch
Thanks
Manjunath
________________________________
From: Mike Artz <mi...@gmail.com>>
Sent: Monday, May 25, 2020 10:50 AM
To: Manjunath Shetty H <ma...@live.com>>
Cc: user <us...@spark.apache.org>>
Subject: Re: Parallelising JDBC reads in spark
Does anything different happened when you set the isolationLevel to do Dirty Reads i.e. "READ_UNCOMMITTED"
On Sun, May 24, 2020 at 7:50 PM Manjunath Shetty H <ma...@live.com>> wrote:
Hi,
We are writing a ETL pipeline using Spark, that fetch the data from SQL server in batch mode (every 15mins). Problem we are facing when we try to parallelising single table reads into multiple tasks without missing any data.
We have tried this,
* Use `ROW_NUMBER` window function in the SQL query
* Then do
*
DataFrame df =
hiveContext
.read()
.jdbc(
<url>,
query,
"row_num",
1,
<upper_limit>,
noOfPartitions,
jdbcOptions);
The problem with this approach is if our tables get updated in between in SQL Server while tasks are still running then the `ROW_NUMBER` will change and we may miss some records.
Any approach to how to fix this issue ? . Any pointers will be helpful
Note: I am on spark 1.6
Thanks
Manjiunath Shetty
Re: Parallelising JDBC reads in spark
Posted by Georg Heiler <ge...@gmail.com>.
Why don't you apply proper change data capture?
This will be more complex though.
Am Mo., 25. Mai 2020 um 07:38 Uhr schrieb Manjunath Shetty H <
manjunathshetty@live.com>:
> Hi Mike,
>
> Thanks for the response.
>
> Even with that flag set data miss can happen right ?. As the fetch is
> based on the last watermark (maximum timestamp of the row that last batch
> job fetched ), Take a scenario like this with table
>
> a : 1
> b : 2
> c : 3
> d : 4
> *f : 6*
> g : 7
> h : 8
> e : 5
>
>
> - a,b,c,d,e get picked by 1 task
> - by the time second task starts, e has been updated, so the row order
> changes
> - As f moves up, it will completely get missed in the fetch
>
>
> Thanks
> Manjunath
>
> ------------------------------
> *From:* Mike Artz <mi...@gmail.com>
> *Sent:* Monday, May 25, 2020 10:50 AM
> *To:* Manjunath Shetty H <ma...@live.com>
> *Cc:* user <us...@spark.apache.org>
> *Subject:* Re: Parallelising JDBC reads in spark
>
> Does anything different happened when you set the isolationLevel to do
> Dirty Reads i.e. "READ_UNCOMMITTED"
>
> On Sun, May 24, 2020 at 7:50 PM Manjunath Shetty H <
> manjunathshetty@live.com> wrote:
>
> Hi,
>
> We are writing a ETL pipeline using Spark, that fetch the data from SQL
> server in batch mode (every 15mins). Problem we are facing when we try to
> parallelising single table reads into multiple tasks without missing any
> data.
>
> We have tried this,
>
>
> - Use `ROW_NUMBER` window function in the SQL query
> - Then do
> -
>
> DataFrame df =
> hiveContext
> .read()
> .jdbc(
> *<url>*,
> query,
> "row_num",
> 1,
> <upper_limit>,
> noOfPartitions,
> jdbcOptions);
>
>
>
> The problem with this approach is if our tables get updated in between in SQL Server while tasks are still running then the `ROW_NUMBER` will change and we may miss some records.
>
>
> Any approach to how to fix this issue ? . Any pointers will be helpful
>
>
> *Note*: I am on spark 1.6
>
>
> Thanks
>
> Manjiunath Shetty
>
>
Re: Parallelising JDBC reads in spark
Posted by Manjunath Shetty H <ma...@live.com>.
Hi Mike,
Thanks for the response.
Even with that flag set data miss can happen right ?. As the fetch is based on the last watermark (maximum timestamp of the row that last batch job fetched ), Take a scenario like this with table
a : 1
b : 2
c : 3
d : 4
f : 6
g : 7
h : 8
e : 5
* a,b,c,d,e get picked by 1 task
* by the time second task starts, e has been updated, so the row order changes
* As f moves up, it will completely get missed in the fetch
Thanks
Manjunath
________________________________
From: Mike Artz <mi...@gmail.com>
Sent: Monday, May 25, 2020 10:50 AM
To: Manjunath Shetty H <ma...@live.com>
Cc: user <us...@spark.apache.org>
Subject: Re: Parallelising JDBC reads in spark
Does anything different happened when you set the isolationLevel to do Dirty Reads i.e. "READ_UNCOMMITTED"
On Sun, May 24, 2020 at 7:50 PM Manjunath Shetty H <ma...@live.com>> wrote:
Hi,
We are writing a ETL pipeline using Spark, that fetch the data from SQL server in batch mode (every 15mins). Problem we are facing when we try to parallelising single table reads into multiple tasks without missing any data.
We have tried this,
* Use `ROW_NUMBER` window function in the SQL query
* Then do
*
DataFrame df =
hiveContext
.read()
.jdbc(
<url>,
query,
"row_num",
1,
<upper_limit>,
noOfPartitions,
jdbcOptions);
The problem with this approach is if our tables get updated in between in SQL Server while tasks are still running then the `ROW_NUMBER` will change and we may miss some records.
Any approach to how to fix this issue ? . Any pointers will be helpful
Note: I am on spark 1.6
Thanks
Manjiunath Shetty