You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by "Davis Zhang (Jira)" <ji...@apache.org> on 2025/01/17 18:23:00 UTC
[jira] [Comment Edited] (HUDI-8692) Re-enable TestHoodieDeltaStreamer#testDeltaStreamerMultiwriterCheckpoint
[ https://issues.apache.org/jira/browse/HUDI-8692?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17914207#comment-17914207 ]
Davis Zhang edited comment on HUDI-8692 at 1/17/25 6:22 PM:
------------------------------------------------------------
h2. *There is a design gap for multi-stream writer*
As of now if I directly run the test it throw exception
{code:java}
java.lang.NumberFormatException: For input string: "topic1,0:10,1:10"
at java.lang.NumberFormatException.forInputString(NumberFormatException.java:65) at java.lang.Long.parseLong(Long.java:589) at java.lang.Long.parseLong(Long.java:631) at org.apache.hudi.utilities.sources.helpers.DFSPathSelector.lambda$getNextFilePathsAndMaxModificationTime$0(DFSPathSelector.java:130) at org.apache.hudi.common.util.Option.map(Option.java:112) at org.apache.hudi.utilities.sources.helpers.DFSPathSelector.getNextFilePathsAndMaxModificationTime(DFSPathSelector.java:130) at org.apache.hudi.utilities.sources.helpers.DFSPathSelector.getNextFilePathsAndMaxModificationTime(DFSPathSelector.java:113) at org.apache.hudi.utilities.sources.ParquetDFSSource.fetchNextBatch(ParquetDFSSource.java:52) at org.apache.hudi.utilities.sources.RowSource.readFromCheckpoint(RowSource.java:61) at org.apache.hudi.utilities.sources.Source.fetchNext(Source.java:139) at org.apache.hudi.utilities.streamer.SourceFormatAdapter.fetchNewDataInAvroFormat(SourceFormatAdapter.java:184) at org.apache.hudi.utilities.streamer.StreamSync.fetchNextBatchFromSource(StreamSync.java:710) at org.apache.hudi.utilities.streamer.StreamSync.fetchFromSourceAndPrepareRecords(StreamSync.java:582) at org.apache.hudi.utilities.streamer.StreamSync.readFromSource(StreamSync.java:554) at org.apache.hudi.utilities.streamer.StreamSync.syncOnce(StreamSync.java:464) at org.apache.hudi.utilities.streamer.HoodieStreamer$StreamSyncService.ingestOnce(HoodieStreamer.java:903) at org.apache.hudi.utilities.ingestion.HoodieIngestionService.startIngestion(HoodieIngestionService.java:72) at org.apache.hudi.common.util.Option.ifPresent(Option.java:101) at org.apache.hudi.utilities.streamer.HoodieStreamer.sync(HoodieStreamer.java:223) at org.apache.hudi.utilities.deltastreamer.TestHoodieDeltaStreamer.testDeltaStreamerMultiwriterCheckpoint(TestHoodieDeltaStreamer.java:2048){code}
The test tries to do:
* parquet DS write to hudi table
{code:java}
"extraMetadata": {
"map": {
"schema": "{\"type\":\"record\",\"name\":\"tripUberRec\",\"fields\":[{\"name\":\"timestamp\",\"type\":\"long\"},{\"name\":\"_row_key\",\"type\":\"string\"},{\"name\":\"rider\",\"type\":\"string\"},{\"name\":\"driver\",\"type\":\"string\"},{\"name\":\"fare\",\"type\":\"double\"},{\"name\":\"_hoodie_is_deleted\",\"type\":\"boolean\"}]}",
"streamer.checkpoint.key.v2": "1737136458028"
}
},{code}
* Json source DS writes to the same table
{code:java}
"extraMetadata": {
"map": {
"schema": "{\"type\":\"record\",\"name\":\"tripUberRec\",\"fields\":[{\"name\":\"timestamp\",\"type\":\"long\"},{\"name\":\"_row_key\",\"type\":\"string\"},{\"name\":\"rider\",\"type\":\"string\"},{\"name\":\"driver\",\"type\":\"string\"},{\"name\":\"fare\",\"type\":\"double\"},{\"name\":\"_hoodie_is_deleted\",\"type\":\"boolean\"}]}",
"streamer.checkpoint.key.v2": "topic1,0:10,1:10"
}
}{code}
* parquet DS write to hudi table again <--- error out here.
It is apparent that there is a design gap of multi-writer - Previously we use a singleton attribute for tracking the checkpoint as there should be only 1 data source all the time. There is only 1 cursor over 1 queue.
Now as we write multiple from data sources, various writers will trash each other as the overload the "streamer.checkpoint.key.v2" for tracking their own state. It is lucky we error out here as different data source have different checkpoint format, preventing from actually writing anything downstream. If they are of the same data source type, it simply means data corruption as from the perspective of 1 data source, it's offset is arbitrarily reset to some random place, leading to either invalid offset or offset pointing to data already processed.
h2. Proposed AI
This is non-trivial. We should track it as a Jira issue and plan for design, dev and test properly in future sprints.
h2. *Design for closing the gap*
To fix this issue, we need a multi-dimension way of tracking checkpoints. Instead of just tracking offset, we need (data source id, offset). 2 questions:
*[Id registry for data sources]*
How to track id for a data source. When do we identify a stream as an existing stream and when do we decide to this is a new stream and id should be assigned accordingly. Do we allow removal of a stream. Especially needs to watch out how this interacts with today's functionality like clean and restart.
*[physical form of storing (data source id, offset) info in instant]*
What's the format inside the instant file that we should keep this piece of information.
h3. *Id registry for data sources*
When user created a delta streamer, 2 things are created:
* A hudi table holding data from the data source.
* A data pipe that piping data from source to the hudi table.
When we do clean and restarts, both are reset.
Now with multi-streamer we are actually expanding the "data pipe" to be multiple ones. It implies the functionality of:
* Adding a data pipe with unique id
* Removing a data pipe given an id
* A registry tracking (id, offset)
So it requires user facing API for the first 2.
{code:java}
// It should be almost the same as today's API of creating delta streamer, just reusing some existing hudi table
AddDataSource(hudi_table_identifier, data_pipe_id, various_data_pipe_configs)
// Once removed, no more writes happens to the data pipe. Ongoing writes are interrupted at a best effort.
RemoveDataSource(hudi_table_identifier, data_pipe_id){code}
Of course, the hudi streamer code needs to accommodate the `data_pipe_id` whenever it tries to access some data pipe specific state - checkpoint is one example. More can come as we investigate the code.
For the third one it will be discussed in a separate section.
{*}physical form of storing (data source id, offset) info in instant{*}{*}{{*}}
Propose: streamer.checkpoint.key.v2.$streamId -> offset
It requires the streamId to conform to [a-z][a-z0-9_]*
was (Author: JIRAUSER305408):
h2. *There is a design gap for multi-stream writer*
As of now if I directly run the test it throw exception
{code:java}
java.lang.NumberFormatException: For input string: "topic1,0:10,1:10"
at java.lang.NumberFormatException.forInputString(NumberFormatException.java:65) at java.lang.Long.parseLong(Long.java:589) at java.lang.Long.parseLong(Long.java:631) at org.apache.hudi.utilities.sources.helpers.DFSPathSelector.lambda$getNextFilePathsAndMaxModificationTime$0(DFSPathSelector.java:130) at org.apache.hudi.common.util.Option.map(Option.java:112) at org.apache.hudi.utilities.sources.helpers.DFSPathSelector.getNextFilePathsAndMaxModificationTime(DFSPathSelector.java:130) at org.apache.hudi.utilities.sources.helpers.DFSPathSelector.getNextFilePathsAndMaxModificationTime(DFSPathSelector.java:113) at org.apache.hudi.utilities.sources.ParquetDFSSource.fetchNextBatch(ParquetDFSSource.java:52) at org.apache.hudi.utilities.sources.RowSource.readFromCheckpoint(RowSource.java:61) at org.apache.hudi.utilities.sources.Source.fetchNext(Source.java:139) at org.apache.hudi.utilities.streamer.SourceFormatAdapter.fetchNewDataInAvroFormat(SourceFormatAdapter.java:184) at org.apache.hudi.utilities.streamer.StreamSync.fetchNextBatchFromSource(StreamSync.java:710) at org.apache.hudi.utilities.streamer.StreamSync.fetchFromSourceAndPrepareRecords(StreamSync.java:582) at org.apache.hudi.utilities.streamer.StreamSync.readFromSource(StreamSync.java:554) at org.apache.hudi.utilities.streamer.StreamSync.syncOnce(StreamSync.java:464) at org.apache.hudi.utilities.streamer.HoodieStreamer$StreamSyncService.ingestOnce(HoodieStreamer.java:903) at org.apache.hudi.utilities.ingestion.HoodieIngestionService.startIngestion(HoodieIngestionService.java:72) at org.apache.hudi.common.util.Option.ifPresent(Option.java:101) at org.apache.hudi.utilities.streamer.HoodieStreamer.sync(HoodieStreamer.java:223) at org.apache.hudi.utilities.deltastreamer.TestHoodieDeltaStreamer.testDeltaStreamerMultiwriterCheckpoint(TestHoodieDeltaStreamer.java:2048){code}
The test tries to do:
* parquet DS write to hudi table
{code:java}
"extraMetadata": {
"map": {
"schema": "{\"type\":\"record\",\"name\":\"tripUberRec\",\"fields\":[{\"name\":\"timestamp\",\"type\":\"long\"},{\"name\":\"_row_key\",\"type\":\"string\"},{\"name\":\"rider\",\"type\":\"string\"},{\"name\":\"driver\",\"type\":\"string\"},{\"name\":\"fare\",\"type\":\"double\"},{\"name\":\"_hoodie_is_deleted\",\"type\":\"boolean\"}]}",
"streamer.checkpoint.key.v2": "1737136458028"
}
},{code}
* Json source DS writes to the same table
{code:java}
"extraMetadata": {
"map": {
"schema": "{\"type\":\"record\",\"name\":\"tripUberRec\",\"fields\":[{\"name\":\"timestamp\",\"type\":\"long\"},{\"name\":\"_row_key\",\"type\":\"string\"},{\"name\":\"rider\",\"type\":\"string\"},{\"name\":\"driver\",\"type\":\"string\"},{\"name\":\"fare\",\"type\":\"double\"},{\"name\":\"_hoodie_is_deleted\",\"type\":\"boolean\"}]}",
"streamer.checkpoint.key.v2": "topic1,0:10,1:10"
}
}{code}
* parquet DS write to hudi table again <--- error out here.
It is apparent that there is a design gap of multi-writer - Previously we use a singleton attribute for tracking the checkpoint as there should be only 1 data source all the time. There is only 1 cursor over 1 queue.
Now as we write multiple from data sources, various writers will trash each other as the overload the "streamer.checkpoint.key.v2" for tracking their own state. It is lucky we error out here as different data source have different checkpoint format, preventing from actually writing anything downstream. If they are of the same data source type, it simply means data corruption as from the perspective of 1 data source, it's offset is arbitrarily reset to some random place, leading to either invalid offset or offset pointing to data already processed.
h2. *Design for closing the gap*
To fix this issue, we need a multi-dimension way of tracking checkpoints. Instead of just tracking offset, we need (data source id, offset). 2 questions:
*[Id registry for data sources]*
How to track id for a data source. When do we identify a stream as an existing stream and when do we decide to this is a new stream and id should be assigned accordingly. Do we allow removal of a stream. Especially needs to watch out how this interacts with today's functionality like clean and restart.
*[physical form of storing (data source id, offset) info in instant]*
What's the format inside the instant file that we should keep this piece of information.
h3. *Id registry for data sources*
When user created a delta streamer, 2 things are created:
* A hudi table holding data from the data source.
* A data pipe that piping data from source to the hudi table.
When we do clean and restarts, both are reset.
Now with multi-streamer we are actually expanding the "data pipe" to be multiple ones. It implies the functionality of:
* Adding a data pipe with unique id
* Removing a data pipe given an id
* A registry tracking (id, offset)
So it requires user facing API for the first 2.
{code:java}
// It should be almost the same as today's API of creating delta streamer, just reusing some existing hudi table
AddDataSource(hudi_table_identifier, data_pipe_id, various_data_pipe_configs)
// Once removed, no more writes happens to the data pipe. Ongoing writes are interrupted at a best effort.
RemoveDataSource(hudi_table_identifier, data_pipe_id){code}
Of course, the hudi streamer code needs to accommodate the `data_pipe_id` whenever it tries to access some data pipe specific state - checkpoint is one example. More can come as we investigate the code.
For the third one it will be discussed in a separate section.
{*}physical form of storing (data source id, offset) info in instant{*}{*}{*}
Propose: streamer.checkpoint.key.v2.$streamId -> offset
It requires the streamId to conform to [a-z][a-z0-9_]*
> Re-enable TestHoodieDeltaStreamer#testDeltaStreamerMultiwriterCheckpoint
> ------------------------------------------------------------------------
>
> Key: HUDI-8692
> URL: https://issues.apache.org/jira/browse/HUDI-8692
> Project: Apache Hudi
> Issue Type: Sub-task
> Components: tests-ci
> Reporter: Y Ethan Guo
> Assignee: Davis Zhang
> Priority: Blocker
> Fix For: 1.0.1
>
>
> Related: HUDI-6609
--
This message was sent by Atlassian Jira
(v8.20.10#820010)