[SPARK-48292][CORE] Revert [SPARK-39195][SQL] Spark OutputCommitCoordinator should abort stage when committed file not consistent with task status #46696

AngersZhuuuu · 2024-05-22T02:59:13Z

What changes were proposed in this pull request?

Revert #36564 According to discuss #36564 (comment)

When spark commit task will commit to committedTaskPath
${outputpath}/_temporary//${appAttempId}/${taskId}
So in #36564 's case, since before #38980, each task's job id's date is not the same, when the task writes data success but fails to send back TaskSuccess RPC, the task rerun will commit to a different committedTaskPath then causing data duplicated.

After #38980, for the same task's different attempts, the TaskId is the same now, when re-run task commit, will commit to the same committedTaskPath, and hadoop CommitProtocol will handle such case then data won't be duplicated.

Note: The taskAttemptPath is not same since in the path contains the taskAttemptId.

Why are the changes needed?

No need anymore

Does this PR introduce any user-facing change?

No

How was this patch tested?

Existed UT

Was this patch authored or co-authored using generative AI tooling?

No

…nator should abort stage when committed file not consistent with task status

AngersZhuuuu · 2024-05-22T02:59:21Z

ping @cloud-fan

cloud-fan · 2024-05-22T11:02:22Z

core/src/test/scala/org/apache/spark/scheduler/OutputCommitCoordinatorIntegrationSuite.scala

    // Regression test for SPARK-10381
-    val e = intercept[SparkException] {
+    failAfter(Span(60, Seconds)) {


shall we still check the error?

Won't throw error after revert...., it can run success.

cloud-fan · 2024-05-22T11:04:03Z

Can we explain this a bit more about why the issue is gone now?

AngersZhuuuu · 2024-05-23T02:17:50Z

Can we explain this a bit more about why the issue is gone now?

Added to pr desc

cloud-fan · 2024-05-23T20:51:48Z

will commit to the same committedTaskPath, and hadoop CommitProtocol will handle such case then data won't be duplicated.

Will we hit file already exist exception in this case?

AngersZhuuuu · 2024-05-24T02:20:16Z

will commit to the same committedTaskPath, and hadoop CommitProtocol will handle such case then data won't be duplicated.

Will we hit file already exist exception in this case?

commitTask will overwrite the existed committedTaskPath , won't throw file already exception.

cloud-fan · 2024-05-24T21:09:30Z

can we also revert #46562 in this PR?

dongjoon-hyun

+1, LGTM. Also, +1 to include the additional revert into here.

cc @viirya

viirya · 2024-05-24T23:05:13Z

Looks good to me.

AngersZhuuuu · 2024-05-29T02:00:49Z

can we also revert #46562 in this PR?

Done

AngersZhuuuu · 2024-05-30T10:28:06Z

GA passed cc @cloud-fan

cloud-fan · 2024-05-30T16:48:19Z

thanks, merging to master!

…inator should abort stage when committed file not consistent with task status ### What changes were proposed in this pull request? Revert apache#36564 According to discuss apache#36564 (comment) When spark commit task will commit to committedTaskPath `${outputpath}/_temporary//${appAttempId}/${taskId}` So in apache#36564 's case, since before apache#38980, each task's job id's date is not the same, when the task writes data success but fails to send back TaskSuccess RPC, the task rerun will commit to a different committedTaskPath then causing data duplicated. After apache#38980, for the same task's different attempts, the TaskId is the same now, when re-run task commit, will commit to the same committedTaskPath, and hadoop CommitProtocol will handle such case then data won't be duplicated. Note: The taskAttemptPath is not same since in the path contains the taskAttemptId. ### Why are the changes needed? No need anymore ### Does this PR introduce _any_ user-facing change? No ### How was this patch tested? Existed UT ### Was this patch authored or co-authored using generative AI tooling? No Closes apache#46696 from AngersZhuuuu/SPARK-48292. Authored-by: Angerszhuuuu <angers.zhu@gmail.com> Signed-off-by: Wenchen Fan <wenchen@databricks.com>

dongjoon-hyun · 2024-07-01T18:15:39Z

Hi, @AngersZhuuuu , @viirya , @cloud-fan .
This seems to cause confusions due to the behavior difference consistently at Apache Spark 3.4 and 3.5. Can we have this backport in old release branches?

viirya · 2024-07-01T18:24:06Z

Hi, @AngersZhuuuu , @viirya , @cloud-fan . This seems to cause confusions due to the behavior difference consistently at Apache Spark 3.4 and 3.5. Can we have this backport in old release branches?

Sounds reasonable to me. Looks like #38980 is also merged into 3.4.

…inator should abort stage when committed file not consistent with task status Revert apache#36564 According to discuss apache#36564 (comment) When spark commit task will commit to committedTaskPath `${outputpath}/_temporary//${appAttempId}/${taskId}` So in apache#36564 's case, since before apache#38980, each task's job id's date is not the same, when the task writes data success but fails to send back TaskSuccess RPC, the task rerun will commit to a different committedTaskPath then causing data duplicated. After apache#38980, for the same task's different attempts, the TaskId is the same now, when re-run task commit, will commit to the same committedTaskPath, and hadoop CommitProtocol will handle such case then data won't be duplicated. Note: The taskAttemptPath is not same since in the path contains the taskAttemptId. No need anymore No Existed UT No Closes apache#46696 from AngersZhuuuu/SPARK-48292. Authored-by: Angerszhuuuu <angers.zhu@gmail.com> Signed-off-by: Wenchen Fan <wenchen@databricks.com>

dongjoon-hyun · 2024-07-01T19:06:48Z

Thank you. Here are backporting PRs.

[SPARK-48292][CORE][3.5] Revert [SPARK-39195][SQL] Spark OutputCommitCoordinator should abort stage when committed file not consistent with task status #47166 (branch-3.5)
[SPARK-48292][CORE][3.4] Revert [SPARK-39195][SQL] Spark OutputCommitCoordinator should abort stage when committed file not consistent with task status #47168 (branch-3.4)

…Coordinator should abort stage when committed file not consistent with task status This is a backport of #46696 ### What changes were proposed in this pull request? Revert #36564 According to discuss #36564 (comment) When spark commit task will commit to committedTaskPath `${outputpath}/_temporary//${appAttempId}/${taskId}` So in #36564 's case, since before #38980, each task's job id's date is not the same, when the task writes data success but fails to send back TaskSuccess RPC, the task rerun will commit to a different committedTaskPath then causing data duplicated. After #38980, for the same task's different attempts, the TaskId is the same now, when re-run task commit, will commit to the same committedTaskPath, and hadoop CommitProtocol will handle such case then data won't be duplicated. Note: The taskAttemptPath is not same since in the path contains the taskAttemptId. ### Why are the changes needed? No need anymore ### Does this PR introduce _any_ user-facing change? No ### How was this patch tested? Existed UT ### Was this patch authored or co-authored using generative AI tooling? No Closes #47166 from dongjoon-hyun/SPARK-48292. Authored-by: Angerszhuuuu <angers.zhu@gmail.com> Signed-off-by: Dongjoon Hyun <dhyun@apple.com>

…Coordinator should abort stage when committed file not consistent with task status This is a backport of #46696 ### What changes were proposed in this pull request? Revert #36564 According to discuss #36564 (comment) When spark commit task will commit to committedTaskPath `${outputpath}/_temporary//${appAttempId}/${taskId}` So in #36564 's case, since before #38980, each task's job id's date is not the same, when the task writes data success but fails to send back TaskSuccess RPC, the task rerun will commit to a different committedTaskPath then causing data duplicated. After #38980, for the same task's different attempts, the TaskId is the same now, when re-run task commit, will commit to the same committedTaskPath, and hadoop CommitProtocol will handle such case then data won't be duplicated. Note: The taskAttemptPath is not same since in the path contains the taskAttemptId. ### Why are the changes needed? No need anymore ### Does this PR introduce _any_ user-facing change? No ### How was this patch tested? Existed UT ### Was this patch authored or co-authored using generative AI tooling? No Closes #47168 from dongjoon-hyun/SPARK-48292-3.4. Authored-by: Angerszhuuuu <angers.zhu@gmail.com> Signed-off-by: Dongjoon Hyun <dhyun@apple.com>

akki · 2024-08-02T09:49:53Z

Hi all

I am facing this issue after upgrading to Spark3.5.1 and wonder if this revert would help me. Does anyone here know that?

Many of our jobs are failing with "Authorized committer" errors and we might have to revert our whole system back to Spark3.3, which would be a lot of work. I am wondering if patching my Spark (to include this commit) would make these failures go away. I would appreciate if anyone who closely understands this diff could confirm (or deny) to understanding.

Thanks!

cloud-fan · 2024-08-02T15:01:08Z

This revert should fix your problem.

dongjoon-hyun · 2024-08-02T15:36:16Z

To @akki , as mentioned by Wenchen, SPARK-48292 fixed it by reverting old patch.

Please try to download and test your case with Apache Spark 3.5.2 RC4.

https://dist.apache.org/repos/dist/dev/spark/v3.5.2-rc4-bin/

akki · 2024-08-04T19:12:42Z

Thanks for the reply both. I'll try applying this patch.

I don't want to audit all the changes included in the 3.5.2RC4 at the moment, so I am leaning towards just reverting the earlier commit for now.

Appreciate the quick responses!

…Coordinator should abort stage when committed file not consistent with task status This is a backport of apache#46696 ### What changes were proposed in this pull request? Revert apache#36564 According to discuss apache#36564 (comment) When spark commit task will commit to committedTaskPath `${outputpath}/_temporary//${appAttempId}/${taskId}` So in apache#36564 's case, since before apache#38980, each task's job id's date is not the same, when the task writes data success but fails to send back TaskSuccess RPC, the task rerun will commit to a different committedTaskPath then causing data duplicated. After apache#38980, for the same task's different attempts, the TaskId is the same now, when re-run task commit, will commit to the same committedTaskPath, and hadoop CommitProtocol will handle such case then data won't be duplicated. Note: The taskAttemptPath is not same since in the path contains the taskAttemptId. ### Why are the changes needed? No need anymore ### Does this PR introduce _any_ user-facing change? No ### How was this patch tested? Existed UT ### Was this patch authored or co-authored using generative AI tooling? No Closes apache#47168 from dongjoon-hyun/SPARK-48292-3.4. Authored-by: Angerszhuuuu <angers.zhu@gmail.com> Signed-off-by: Dongjoon Hyun <dhyun@apple.com>

lihao712 · 2024-09-19T12:47:19Z

will commit to the same committedTaskPath, and hadoop CommitProtocol will handle such case then data won't be duplicated.

Will we hit file already exist exception in this case?

commitTask will overwrite the existed committedTaskPath , won't throw file already exception.

As far as I know, after Hadoop 2.7, the algorithm for checking the task commit path is version 2, and the version 1 implementation performs poorly in practice. For tasks with a large number of files, under the algorithm version 2, how can we ensure that there won't be two task commit files for the same partition existing simultaneously in the final directory?

dongjoon-hyun · 2024-09-19T15:29:17Z

To @lihao712 , Apache Spark (3.0.2+) uses version 1 by default (via SPARK-33019) due to the correctness issue of version 1 (MAPREDUCE-7282).

As far as I know, after Hadoop 2.7, the algorithm for checking the task commit path is version 2, and the version 1 implementation performs poorly in practice. For tasks with a large number of files, under the algorithm version 2, how can we ensure that there won't be two task commit files for the same partition existing simultaneously in the final directory?

[SPARK-33019][CORE] Use spark.hadoop.mapreduce.fileoutputcommitter.algorithm.version=1 by default #29895

lihao712 · 2024-09-20T02:19:35Z

To @lihao712 , Apache Spark (3.0.2+) uses version 1 by default (via SPARK-33019) due to the correctness issue of version 1 (MAPREDUCE-7282).

As far as I know, after Hadoop 2.7, the algorithm for checking the task commit path is version 2, and the version 1 implementation performs poorly in practice. For tasks with a large number of files, under the algorithm version 2, how can we ensure that there won't be two task commit files for the same partition existing simultaneously in the final directory?

[SPARK-33019][CORE] Use spark.hadoop.mapreduce.fileoutputcommitter.algorithm.version=1 by default #29895

However, the performance of Algorithm 1 is significantly worse than that of Algorithm 2. Have you tested the performance of both algorithms in scenarios where there are a large number of files produced in the partition? Additionally, has Hadoop made any optimizations to improve the performance of Algorithm 1?

mridulm · 2024-09-20T07:01:51Z

@lihao712, as @dongjoon-hyun mentioned above - v1 is used given correctness issue of v2.
Correctness takes precedence over performance

[SPARK-48292[CORE] Revert [SPARK-39195][SQL] Spark OutputCommitCoordi…

7e49844

…nator should abort stage when committed file not consistent with task status

github-actions bot added the CORE label May 22, 2024

cloud-fan reviewed May 22, 2024

View reviewed changes

cloud-fan approved these changes May 24, 2024

View reviewed changes

dongjoon-hyun approved these changes May 24, 2024

View reviewed changes

viirya approved these changes May 24, 2024

View reviewed changes

Update ParquetIOSuite.scala

0a426d2

github-actions bot added the SQL label May 29, 2024

cloud-fan approved these changes May 29, 2024

View reviewed changes

cloud-fan closed this in f68d761 May 30, 2024

dongjoon-hyun mentioned this pull request Jul 1, 2024

[SPARK-48292][CORE][3.5] Revert [SPARK-39195][SQL] Spark OutputCommitCoordinator should abort stage when committed file not consistent with task status #47166

Closed

dongjoon-hyun mentioned this pull request Jul 1, 2024

[SPARK-48292][CORE][3.4] Revert [SPARK-39195][SQL] Spark OutputCommitCoordinator should abort stage when committed file not consistent with task status #47168

Closed

dongjoon-hyun mentioned this pull request Jul 12, 2024

[SPARK-39195][SQL] Spark OutputCommitCoordinator should abort stage when committed file not consistent with task status #36564

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[SPARK-48292][CORE] Revert [SPARK-39195][SQL] Spark OutputCommitCoordinator should abort stage when committed file not consistent with task status #46696

[SPARK-48292][CORE] Revert [SPARK-39195][SQL] Spark OutputCommitCoordinator should abort stage when committed file not consistent with task status #46696

AngersZhuuuu commented May 22, 2024 •

edited

Loading

AngersZhuuuu commented May 22, 2024

cloud-fan May 22, 2024

AngersZhuuuu May 23, 2024

cloud-fan commented May 22, 2024

AngersZhuuuu commented May 23, 2024

cloud-fan commented May 23, 2024

AngersZhuuuu commented May 24, 2024

cloud-fan commented May 24, 2024

dongjoon-hyun left a comment

viirya commented May 24, 2024

AngersZhuuuu commented May 29, 2024

AngersZhuuuu commented May 30, 2024

cloud-fan commented May 30, 2024

dongjoon-hyun commented Jul 1, 2024

viirya commented Jul 1, 2024

dongjoon-hyun commented Jul 1, 2024

akki commented Aug 2, 2024 •

edited

Loading

cloud-fan commented Aug 2, 2024

dongjoon-hyun commented Aug 2, 2024

akki commented Aug 4, 2024

lihao712 commented Sep 19, 2024

dongjoon-hyun commented Sep 19, 2024 •

edited

Loading

lihao712 commented Sep 20, 2024

mridulm commented Sep 20, 2024

[SPARK-48292][CORE] Revert [SPARK-39195][SQL] Spark OutputCommitCoordinator should abort stage when committed file not consistent with task status #46696

[SPARK-48292][CORE] Revert [SPARK-39195][SQL] Spark OutputCommitCoordinator should abort stage when committed file not consistent with task status #46696

Conversation

AngersZhuuuu commented May 22, 2024 • edited Loading

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

Was this patch authored or co-authored using generative AI tooling?

AngersZhuuuu commented May 22, 2024

cloud-fan May 22, 2024

Choose a reason for hiding this comment

AngersZhuuuu May 23, 2024

Choose a reason for hiding this comment

cloud-fan commented May 22, 2024

AngersZhuuuu commented May 23, 2024

cloud-fan commented May 23, 2024

AngersZhuuuu commented May 24, 2024

cloud-fan commented May 24, 2024

dongjoon-hyun left a comment

Choose a reason for hiding this comment

viirya commented May 24, 2024

AngersZhuuuu commented May 29, 2024

AngersZhuuuu commented May 30, 2024

cloud-fan commented May 30, 2024

dongjoon-hyun commented Jul 1, 2024

viirya commented Jul 1, 2024

dongjoon-hyun commented Jul 1, 2024

akki commented Aug 2, 2024 • edited Loading

cloud-fan commented Aug 2, 2024

dongjoon-hyun commented Aug 2, 2024

akki commented Aug 4, 2024

lihao712 commented Sep 19, 2024

dongjoon-hyun commented Sep 19, 2024 • edited Loading

lihao712 commented Sep 20, 2024

mridulm commented Sep 20, 2024

AngersZhuuuu commented May 22, 2024 •

edited

Loading

akki commented Aug 2, 2024 •

edited

Loading

dongjoon-hyun commented Sep 19, 2024 •

edited

Loading