apache / apache/hudi

[SUPPORT] When using Deltasteamer JdbcSource to extract data, there are issues with data loss and slow query of source side data

Open
#9,471 7 comments 0 reactions 0 assignees View on GitHub
area:ingest issue:data-loss priority:critical status:triaged
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
![855a024e8205c543de322aba36ee924](https://github.com/apache/hudi/assets/65161474/50301520-bac6-4ada-897b-63d93c79b3b1)
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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.