[SUPPORT] FlinkCDC+Hudi0.10 failure when delete or update history data but success on currentdate data
- Dominant language
- Java
- Stars
- 6.2k
- Forks
- 2.5k
- Avg merge
- 2d 8h
- Merged PRs (30d)
- 111
Description
I recently made a small demo about CDC,
With the Flink CDC capture Mysql data changes and Sink to Hudi, synchronized to the hive.
But when I update, or delete data,
it failure when delete or update history data but success on currentdate data
**Environment Description**
* Hudi version :0.10
* Hive version :2.1.1
* flink version : 1.13
* Hadoop version : 3.0.0
* Storage (HDFS/S3/GCS..) :hdfs
* Running on Docker? (yes/no) : no
**To Reproduce**
mysql has a table users withs 2 rows
id name ts
8 scala 2022-10-12 07:13:40.000
24 flink1.13 flink1.13 **2022-11-10 10:41:18.000**
then
1.update users set name='python' wherer id=8
2. delete from users wherer id=8
3.update users set name='flink1.13bug' wherer id=24
**Expected behavior**
Finished step 3, as expected hive synchronous table can see id = 8 records deleted and id = 24 records be updated
But id = 8 records are still in hive synchronous table and its name still eq 'scala ',so update invalid too.
but id = 24 success
**Additional context**
then I open flink sql client ,saw the sink table changes
+----+----------------------+--------------------------------+-------------------------+-------------------------+--------------------------------+
| op | id | name | birthday | ts | partition |
+----+----------------------+--------------------------------+-------------------------+-------------------------+--------------------------------+
| +I | 8 | scala | 2022-10-12 15:13:40.000 | 2022-10-12 15:13:40.000 | 20221012 |
| -U | 8 | scala | 2022-10-12 07:13:40.000 | 2022-10-12 07:13:40.000 | 20221012 |
| +U | 8 | python | 2022-10-12 07:13:40.000 | 2022-10-12 07:13:40.000 | 20221012 |
**| +I | 8 | scala |** 2022-10-12 15:13:40.000 | 2022-10-12 15:13:40.000 | 20221012 |
| -D | 8 | python | 2022-10-12 07:13:40.000 | 2022-10-12 07:13:40.000 | 20221012 |
**| +I | 8 | scala |** 2022-10-12 15:13:40.000 | 2022-10-12 15:13:40.000 | 20221012 |
Problem is very strange,
in each -U. + U, - D after operation, there will be + I, to insert old data to the sink list
I Don't know why, MySQL data has been modified,There should not be rolled back
the current data 20221110(id=24) data OK, it insert(+I) a new name into table
| -U | 24 | flink1.13 | 2022-11-10 09:37:14.000 | 2022-11-10 09:37:14.000 | 20221110 |
| +U | 24 | flink1.13bug | 2022-11-10 09:37:14.000 | 2022-11-10 09:37:14.000 | 20221110 |
| +I | 24 | flink1.13bug | 2022-11-10 09:37:14.000 | 2022-11-10 09:37:14.000 | 20221110 |
This kind of problem did not show when sink to ES, kafka .
Is this a hudi problem?, hope to get more information
-----------sql ---------------
--SOURCE TABLE
CREATE TABLE mysql_users (
id BIGINT PRIMARY KEY NOT ENFORCED ,
name STRING,
birthday TIMESTAMP(3),
ts TIMESTAMP(3)
)
with
('connector'='mysql-cdc'
,'hostname'='xxxx'
,'port'='3306'
,'username'='xxx'
,'password'='xxx'
,'debezium.snapshot.mode'='initial'
,'database-name'='flink_cdc_test'
,'table-name'='users')
--SINK TABLE
CREATE TABLE users_mor_sync2hive01(
id bigint ,
name string,
birthday TIMESTAMP(3),
ts TIMESTAMP(3),
`partition` VARCHAR(20),
primary key(id) not enforced
)partitioned by (partition string)
with(
'connector'='hudi',
'path'= 'hdfs://nameservice1/hudidatas/hudi-warehouse/users_mor_sync2hive01'
, 'hoodie.datasource.write.recordkey.field'= 'id'
, 'write.precombine.field'= 'ts'
, 'write.tasks'= '1'
, 'compaction.tasks'= '1'
, 'write.rate.limit'= '2000'
, 'table.type'= 'MERGE_ON_READ'
, 'compaction.async.enabled'= 'true'
, 'compaction.trigger.strategy'= 'num_commits'
, 'compaction.delta_commits'= '1'
, 'changelog.enabled'= 'true'
, 'read.streaming.enabled'= 'true'
, 'read.streaming.check-interval'= '3'
, 'hive_sync.enable'= 'true'
, 'hive_sync.mode'= 'hms'
, 'hive_sync.metastore.uris'= 'thrift://XXXX:9083'
-- , 'hive_sync.jdbc_url'= 'jdbc:hive2://hadoop:10000'
, 'hive_sync.table'= 'users_mor_sync2hive01'
, 'hive_sync.db'= 'hudi2hive'
, 'hive_sync.username'= ''
, 'hive_sync.password'= ''
, 'hive_sync.support_timestamp'= 'true'
);
INSERT INTO users_mor_sync2hive01 SELECT *, DATE_FORMAT(birthday, 'yyyyMMdd') FROM mysql_users;
Contributor guide
No contributing guide indexed for this repository
Research direction
Start with the provided MySQL CDC source and Hudi sink table definitions, especially the record-key, precombine, partitioning, and merge-on-read options. Reproduce the update and delete sequence for the historical and current-date rows, then compare the sink and Hive results; done means the cause of the historical-row reinserts and the expected final state are established.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- mysql
- Domain
- data-engineering, databases, stream-processing
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Stale
- Clarity
- Needs clarification
- Newbie friendliness
- 25/100