Skip to content

Commit

Permalink
[SPARK-41407][SQL][FOLLOW-UP] Use string jobTrackerID for FileFormatW…
Browse files Browse the repository at this point in the history
…riter.executeTask

### What changes were proposed in this pull request?

This PR is a followup of #38939 that fixes a logical conflict during merging PRs, see #38980 and #38939.

### Why are the changes needed?

To recover the broken build.

### Does this PR introduce _any_ user-facing change?

No, dev-only.

### How was this patch tested?

Manually tested:

```
 ./build/sbt -Phive clean package
```

Closes #39194 from HyukjinKwon/SPARK-41407.

Authored-by: Hyukjin Kwon <gurwls223@apache.org>
Signed-off-by: Hyukjin Kwon <gurwls223@apache.org>
  • Loading branch information
HyukjinKwon committed Dec 23, 2022
1 parent e1af3a9 commit 3d5af03
Showing 1 changed file with 3 additions and 3 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@ package org.apache.spark.sql.execution.datasources
import java.util.Date

import org.apache.spark.{SparkException, TaskContext}
import org.apache.spark.internal.io.FileCommitProtocol
import org.apache.spark.internal.io.{FileCommitProtocol, SparkHadoopWriterUtils}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.expressions.Attribute
Expand Down Expand Up @@ -72,15 +72,15 @@ case class WriteFilesExec(child: SparkPlan) extends UnaryExecNode {
val concurrentOutputWriterSpec = writeFilesSpec.concurrentOutputWriterSpecFunc(child)
val description = writeFilesSpec.description
val committer = writeFilesSpec.committer
val jobIdInstant = new Date().getTime
val jobTrackerID = SparkHadoopWriterUtils.createJobTrackerID(new Date())
rddWithNonEmptyPartitions.mapPartitionsInternal { iterator =>
val sparkStageId = TaskContext.get().stageId()
val sparkPartitionId = TaskContext.get().partitionId()
val sparkAttemptNumber = TaskContext.get().taskAttemptId().toInt & Int.MaxValue

val ret = FileFormatWriter.executeTask(
description,
jobIdInstant,
jobTrackerID,
sparkStageId,
sparkPartitionId,
sparkAttemptNumber,
Expand Down

0 comments on commit 3d5af03

Please sign in to comment.