apache / apache/hudi

[SUPPORT] FlinkCDC+Hudi0.10 failure when delete or update history data but success on currentdate data

Open
#7,177 2 comments 0 reactions 0 assignees View on GitHub
engine:flink priority:high
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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.