[SUPPORT] When using Deltasteamer JdbcSource to extract data, there are issues with data loss and slow query of source side data
- Dominant language
- Java
- Stars
- 6.2k
- Forks
- 2.5k
- Avg merge
- 2d 8h
- Merged PRs (30d)
- 111
Description
**Describe the problem you faced**
Q1:
Assuming the source table order table has a total data volume of 5 million. Synchronize using deltasteamer [JdbcSource](https://github.com/apache/hudi/blob/ba5ab8ca46863a67023e7172fb16a9a36d3b5acb/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/JdbcSource.java#L183)
Hudi conf:
` --hoodie-conf hoodie.deltastreamer.jdbc.incr.pull=true`
`--hoodie-conf hoodie.deltastreamer.jdbc.table.incr.column.name=update_date`
`--source-limit 100000`
`--continuous`
When deltasteamer synchronizes to 40w data, the current lastCheckpoint=2023-08-17 14:55 0:00:00 So the SQL for
[incrementalFetch ](https://github.com/apache/hudi/blob/ba5ab8ca46863a67023e7172fb16a9a36d3b5acb/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/JdbcSource.java#L206)Method to query source data is:
`select (select * from order where update_date>"2023-08-17 14:55 0:00:00" order by update_date limit 100000) rdbms_table`
Assuming that there is 200000 data in the updateDate field of my order table, which is equal to "2023-08-17 14:55 1:00:000" will
only obtain 100000 rows of data due to sourceLimit=100000, and will also lose 100000 rows of data.
Q2:
Why are these two parameters set?
`--source-limit 100000`
`--continuous`
If these two parameters are not set, [JdbcSource](https://github.com/apache/hudi/blob/ba5ab8ca46863a67023e7172fb16a9a36d3b5acb/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/JdbcSource.java#L183) will perform a full query on the source table.
For tables with a large amount of data, a full query will bring high load or even downtime to the database server
**To Reproduce**
Steps to reproduce the behavior:
1. Prepare Mysql Table Data:
```
create table test.deltasteamer_jdbc_source(
id bigint(11) primary key,
name varchar(50),
incrColumn bigint(11) not null
);
insert into test.deltasteamer_jdbc_source values (1,"Hadoop", 1);
insert into test.deltasteamer_jdbc_source values (2,"Hbase", 1);
insert into test.deltasteamer_jdbc_source values (3,"Hudi", 1);
insert into test.deltasteamer_jdbc_source values (4,"Spark", 1);
insert into test.deltasteamer_jdbc_source values (5,"Flink", 2);
```
2.Prepare HudiConf:
```
hoodie.datasource.write.recordkey.field=id
hoodie.datasource.write.keygenerator.class=org.apache.hudi.keygen.NonpartitionedKeyGenerator
hoodie.deltastreamer.jdbc.incr.pull=true
hoodie.deltastreamer.jdbc.table.incr.column.name=incrColumn
hoodie.deltastreamer.schemaprovider.source.schema.jdbc.nullable=true
hoodie.deltastreamer.jdbc.url=jdbc:mysql://localhost3306/test?serverTimezone=Asia/Shanghai
hoodie.deltastreamer.jdbc.user=root
hoodie.deltastreamer.jdbc.password=
hoodie.deltastreamer.jdbc.driver.class=com.mysql.cj.jdbc.Driver
hoodie.deltastreamer.jdbc.table.name=deltasteamer_jdbc_source
hoodie.deltastreamer.transformer.sql=SELECT * FROM t
```
3.Execute Command
```
spark-submit --master local --class org.apache.hudi.utilities.deltastreamer.HoodieDeltaStreamer --driver-memory 1G --num-executors 1 --executor-cores 1 --executor-memory 1G ./hudi-utilities-bundle_2.12-0.13.1.jar --table-type COPY_ON_WRITE \
--props /tmp/test_jdbc_source.properties \
--source-class org.apache.hudi.utilities.sources.JdbcSource \
--source-limit 3 \
--source-ordering-field incrColumn \
--continuous \
--target-base-path /user/hive/warehouse/test.db/deltasteamer_jdbc_source \
--target-table deltasteamer_jdbc_source \
--transformer-class org.apache.hudi.utilities.transform.SqlQueryBasedTransformer
```
4.Create HudiTable and Query HudiTable Data

Affected by sourcelimit, data loss for name=Spark
**Expected behavior**
A clear and concise description of what you expected to happen.
**Environment Description**
* Hudi version :0.13.1
* Spark version :3.2.1
* Hive version :3.1.3
* Hadoop version :3.3.3
* Storage (HDFS/S3/GCS..) :HDFS
* Running on Docker? (yes/no) :no
**Additional context**
Add any other context about the problem here.
**Stacktrace**
```Add the stacktrace of the error.```
Contributor guide
No contributing guide indexed for this repository
Research direction
Start with hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/JdbcSource.java, especially incrementalFetch and the linked query construction. Run the provided MySQL reproduction with --source-limit 3 and --continuous, then compare the source rows with the Hudi table and observe the generated query. Done means the reported incremental extraction behavior is explained and corrected without requiring an unbounded full-table query.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java, mysql, spark
- Domain
- data-engineering, databases, stream-processing
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Quiet
- Clarity
- Needs clarification
- Newbie friendliness
- 45/100