Skip to content

[SPARK-23559][SS] Add epoch ID to DataWriterFactory. #20710

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
jose-torres wants to merge 11 commits into from
Closed

[SPARK-23559][SS] Add epoch ID to DataWriterFactory. #20710

Changes from 1 commit
Commits
Show all changes
11 commits
Select commit Hold shift + click to select a range
42bca60
change api
jose-torres Feb 21, 2018
84cfa21
add docs
jose-torres Feb 21, 2018
a18a57b
don't use epochId in batch sources
jose-torres Feb 22, 2018
b2ee7f3
change comment
jose-torres Feb 22, 2018
55b38db
fix docs
jose-torres Mar 2, 2018
544eb1b
fix docs
jose-torres Mar 2, 2018
f5948e8
clarify DataWriter lifecycle
jose-torres Mar 3, 2018
4588616
fix foreach
jose-torres Mar 3, 2018
79495b1
various doc fixes
jose-torres Mar 3, 2018
9fb74e2
describe what's an epoch
jose-torres Mar 3, 2018
215c225
clarify retry docs
jose-torres Mar 3, 2018
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
don't use epochId in batch sources
  • Loading branch information
@jose-torres
jose-torres committed Mar 2, 2018
commit a18a57b7aeaf6c0f1bca03e608566629f57cf956
4 changes: 2 additions & 2 deletions sql/core/src/test/scala/org/apache/spark/sql/sources/v2/SimpleWritableDataSource.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -212,7 +212,7 @@ class SimpleCSVDataWriterFactory(path: String, jobId: String, conf: Serializable
attemptNumber: Int,
epochId: Long): DataWriter[Row] = {
val jobPath = new Path(new Path(path, "_temporary"), jobId)
val filePath = new Path(jobPath, s"$jobId-$partitionId-$attemptNumber-$epochId")
val filePath = new Path(jobPath, s"$jobId-$partitionId-$attemptNumber")
val fs = filePath.getFileSystem(conf.value)
new SimpleCSVDataWriter(fs, filePath)
}
Expand Down Expand Up @@ -248,7 +248,7 @@ class InternalRowCSVDataWriterFactory(path: String, jobId: String, conf: Seriali
attemptNumber: Int,
epochId: Long): DataWriter[InternalRow] = {
val jobPath = new Path(new Path(path, "_temporary"), jobId)
val filePath = new Path(jobPath, s"$jobId-$partitionId-$attemptNumber-$epochId")
val filePath = new Path(jobPath, s"$jobId-$partitionId-$attemptNumber")
val fs = filePath.getFileSystem(conf.value)
new InternalRowCSVDataWriter(fs, filePath)
}
Expand Down