zdl11111 opened a new issue, #9940:
URL: https://github.com/apache/hudi/issues/9940
Hi, I am trying to use Flink Hudi to build a streaming Data Lake, but it
seems like HUDI cannot handle the rowKind correctly. I will appreciate it if
anyone can support any help!
1. Prepare a segment of CDC data in debezium-json format in advance
2. Create a table from the Flink SQL Client to read the CDC data files:
CREATE TABLE debezium_source(
id INT NOT NULL,
ts BIGINT,
name STRING,
description STRING,
weight DOUBLE
) WITH (
'connector' = 'filesystem',
'path' = '/tmp/source.data',
'format' = 'debezium-json'
);
3. Execute SELECT to observe the results. We can see that there are a total
of 20 records with some UPDATEs in the middle. The last message is DELETE:
select * from debezium_source;
| op | id | ts | name |
description | weight |
| +I | 101 | 1000 | scooter |
Small 2-wheel scooter | 3.140000104904175 |
| +I | 102 | 2000 | car battery |
12V car battery | 8.100000381469727 |
| +I | 103 | 3000 | 12-pack drill bits |
12-pack of drill bits with ... | 0.800000011920929 |
| +I | 104 | 4000 | hammer |
12oz carpenter's hammer | 0.75 |
| +I | 105 | 5000 | hammer |
14oz carpenter's hammer | 0.875 |
| +I | 106 | 6000 | hammer |
16oz carpenter's hammer | 1.0 |
| +I | 107 | 7000 | rocks |
box of assorted rocks | 5.300000190734863 |
| +I | 108 | 8000 | jacket |
water resistent black wind ... | 0.10000000149011612 |
| +I | 109 | 9000 | spare tire |
24 inch spare tire | 22.200000762939453 |
| -U | 106 | 6000 | hammer |
16oz carpenter's hammer | 1.0 |
| +U | 106 | 10000 | hammer |
18oz carpenter hammer | 1.0 |
| -U | 107 | 7000 | rocks |
box of assorted rocks | 5.300000190734863 |
| +U | 107 | 11000 | rocks |
box of assorted rocks | 5.099999904632568 |
| +I | 110 | 12000 | jacket |
water resistent white wind ... | 0.20000000298023224 |
| +I | 111 | 13000 | scooter |
Big 2-wheel scooter | 5.179999828338623 |
| -U | 110 | 12000 | jacket |
water resistent white wind ... | 0.20000000298023224 |
| +U | 110 | 14000 | jacket |
new water resistent white w... | 0.5 |
| -U | 111 | 13000 | scooter |
Big 2-wheel scooter | 5.179999828338623 |
| +U | 111 | 15000 | scooter |
Big 2-wheel scooter | 5.170000076293945 |
| -D | 111 | 16000 | scooter |
Big 2-wheel scooter | 5.170000076293945 |
Received a total of 20 rows
4.Create a Hudi table:
CREATE TABLE hoodie_table(
id INT NOT NULL PRIMARY KEY NOT ENFORCED,
ts BIGINT,
name STRING,
description STRING,
weight DOUBLE
) WITH (
'connector' = 'hudi',
'path' = 'hdfs://hudi/flink/hoodie_table',
'write.precombine.field' = 'ts',
'table.type' = 'MERGE_ON_READ',
'changelog.enabled' = 'true',
'read.streaming.skip_compaction'='true',
'compaction.async.enabled' = 'true'
);
5. insert into hoodie_table select * from debezium_source;
6. select * from hoodie_table/*+ OPTIONS('read.streaming.enabled'='true')*/;
| op | id | ts | name |
description | weight |
+----+-------------+----------------------+--------------------------------+--------------------------------+--------------------------------+
| +I | 101 | 1000 | scooter |
Small 2-wheel scooter | 3.140000104904175 |
| +I | 102 | 2000 | car battery |
12V car battery | 8.100000381469727 |
| +I | 103 | 3000 | 12-pack drill bits |
12-pack of drill bits with ... | 0.800000011920929 |
| +I | 104 | 4000 | hammer |
12oz carpenter's hammer | 0.75 |
| +I | 105 | 5000 | hammer |
14oz carpenter's hammer | 0.875 |
| +I | 106 | 6000 | hammer |
16oz carpenter's hammer | 1.0 |
| +I | 107 | 7000 | rocks |
box of assorted rocks | 5.300000190734863 |
| +I | 108 | 8000 | jacket |
water resistent black wind ... | 0.10000000149011612 |
| +I | 109 | 9000 | spare tire |
24 inch spare tire | 22.200000762939453 |
| -D | 106 | 6000 | hammer |
16oz carpenter's hammer | 1.0 |
| +I | 106 | 10000 | hammer |
18oz carpenter hammer | 1.0 |
| -D | 107 | 7000 | rocks |
box of assorted rocks | 5.300000190734863 |
| +I | 107 | 11000 | rocks |
box of assorted rocks | 5.099999904632568 |
| +I | 110 | 12000 | jacket |
water resistent white wind ... | 0.20000000298023224 |
| +I | 111 | 13000 | scooter |
Big 2-wheel scooter | 5.179999828338623 |
| -D | 110 | 12000 | jacket |
water resistent white wind ... | 0.20000000298023224 |
| +I | 110 | 14000 | jacket |
new water resistent white w... | 0.5 |
| -D | 111 | 13000 | scooter |
Big 2-wheel scooter | 5.179999828338623 |
| +I | 111 | 15000 | scooter |
Big 2-wheel scooter | 5.170000076293945 |
As we can see, HUDI processes the -U/+U operation as -D and +I, and ignores
the D operation, so we can still get the data which id is '111'(it should be
deleted).
7. select * from hoodie_table;
| op | id | ts | name |
description | weight |
+----+-------------+----------------------+--------------------------------+--------------------------------+--------------------------------+
| +I | 110 | 14000 | jacket |
new water resistent white w... | 0.5 |
**| +I | 111 | 15000 | scooter
| Big 2-wheel scooter | 5.170000076293945 |**
| +I | 101 | 1000 | scooter |
Small 2-wheel scooter | 3.140000104904175 |
| +I | 102 | 2000 | car battery |
12V car battery | 8.100000381469727 |
| +I | 103 | 3000 | 12-pack drill bits |
12-pack of drill bits with ... | 0.800000011920929 |
| +I | 104 | 4000 | hammer |
12oz carpenter's hammer | 0.75 |
| +I | 105 | 5000 | hammer |
14oz carpenter's hammer | 0.875 |
| +I | 106 | 10000 | hammer |
18oz carpenter hammer | 1.0 |
| +I | 107 | 11000 | rocks |
box of assorted rocks | 5.099999904632568 |
| +I | 108 | 8000 | jacket |
water resistent black wind ... | 0.10000000149011612 |
| +I | 109 | 9000 | spare tire |
24 inch spare tire | 22.200000762939453 |
**Environment Description**
* Hudi version : 0.13.1
* Flink version : 1.14.3
* Hive version :
* Hadoop version :
* Storage (HDFS/S3/GCS..) : HDFS
* Running on Docker? (yes/no) :
**Additional context**
Add any other context about the problem here.
**Stacktrace**
```Add the stacktrace of the error.```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
