Skip to content

[Spark-21996][SQL] read files with space in name for streaming #19247

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
wants to merge 9 commits into from
Closed

[Spark-21996][SQL] read files with space in name for streaming #19247

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
38901ef
add test to reproduce
xysun Sep 15, 2017
4f5979a
use URI.getPath instead of toString
xysun Sep 15, 2017
435f42b
Merge branch 'master' into SPARK-21996
xysun Jan 15, 2018
2542014
fix per comment
xysun Jan 15, 2018
f07ccbe
pass in file name prefix for AddTextFileData instead of a new class
xysun Jan 17, 2018
7342d6c
test case for text files generated by file sink
xysun Jan 17, 2018
6e4f809
fix scalastyle
xysun Jan 17, 2018
04c2b14
more scalastyle fix
xysun Jan 17, 2018
10106b3
one more scalastyle..
xysun Jan 17, 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
2 changes: 1 addition & 1 deletion sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSource.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -166,7 +166,7 @@ class FileStreamSource(
val newDataSource =
DataSource(
sparkSession,
paths = files.map(_.path),
paths = files.map(f => new Path(new URI(f.path)).toString),
userSpecifiedSchema = Some(schema),
partitionColumns = partitionColumns,
className = fileFormatClassName,
Expand Down
50 changes: 48 additions & 2 deletions sql/core/src/test/scala/org/apache/spark/sql/streaming/FileStreamSourceSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -74,11 +74,11 @@ abstract class FileStreamSourceTest
protected def addData(source: FileStreamSource): Unit
}

case class AddTextFileData(content: String, src: File, tmp: File)
case class AddTextFileData(content: String, src: File, tmp: File, tmpFilePrefix: String = "text")
extends AddFileData {

override def addData(source: FileStreamSource): Unit = {
val tempFile = Utils.tempFileWith(new File(tmp, "text"))
val tempFile = Utils.tempFileWith(new File(tmp, tmpFilePrefix))
val finalFile = new File(src, tempFile.getName)
src.mkdirs()
require(stringToFile(tempFile, content).renameTo(finalFile))
Expand Down Expand Up @@ -408,6 +408,52 @@ class FileStreamSourceSuite extends FileStreamSourceTest {
}
}

test("SPARK-21996 read from text files -- file name has space") {
Copy link
Contributor

@mgaido91 mgaido91 Jan 15, 2018

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

can we run the same test also for the other input format, ie. parquet, orc, ... ?

Copy link
Member

@zsxwing zsxwing Jan 16, 2018

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

this test should be enough. The issue is in file stream source.

Copy link
Contributor

@mgaido91 mgaido91 Jan 16, 2018

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

yes, for this PR it is, but it would be great if we can ensure that all the data sources have the same behavior... Maybe we can do this is another PR if you think it is better

Copy link
Member

@zsxwing zsxwing Jan 16, 2018

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Not. I meant it's not an issue of file formats. There are not some special codes in file stream source. If there should be any tests for such issue, they should be inside file format tests.

withTempDirs { case (src, tmp) =>
val textStream = createFileStream("text", src.getCanonicalPath)
val filtered = textStream.filter($"value" contains "keep")

testStream(filtered)(
AddTextFileData("drop1\nkeep2\nkeep3", src, tmp, "text text"),
CheckAnswer("keep2", "keep3")
)
}
}

test("SPARK-21996 read from text files generated by file sink -- file name has space") {
val testTableName = "FileStreamSourceTest"
withTable(testTableName) {
withTempDirs { case (src, checkpoint) =>
val output = new File(src, "text text")
val inputData = MemoryStream[String]
val ds = inputData.toDS()

val query = ds.writeStream
.option("checkpointLocation", checkpoint.getCanonicalPath)
.format("text")
.start(output.getCanonicalPath)

try {
inputData.addData("foo")
failAfter(streamingTimeout) {
query.processAllAvailable()
}
} finally {
query.stop()
}

val df2 = spark.readStream.format("text").load(output.getCanonicalPath)
val query2 = df2.writeStream.format("memory").queryName(testTableName).start()
try {
query2.processAllAvailable()
checkDatasetUnorderly(spark.table(testTableName).as[String], "foo")
} finally {
query2.stop()
}
}
}
}

test("read from textfile") {
withTempDirs { case (src, tmp) =>
val textStream = spark.readStream.textFile(src.getCanonicalPath)
Expand Down