[SPARK-17165][SQL] FileStreamSource should not track the list of seen files indefinitely by petermaxlee · Pull Request #14728 · apache/spark

petermaxlee · 2016-08-20T06:40:13Z

What changes were proposed in this pull request?

Before this change, FileStreamSource uses an in-memory hash set to track the list of files processed by the engine. The list can grow indefinitely, leading to OOM or overflow of the hash set.

This patch introduces a new user-defined option called "maxFileAge", default to 24 hours. If a file is older than this age, FileStreamSource will purge it from the in-memory map that was used to track the list of files that have been processed.

How was this patch tested?

Added unit tests for the underlying utility, and also added an end-to-end test to validate the purge in FileStreamSourceSuite. Also verified the new test cases would fail when the timeout was set to a very large number.

… files indefinitely

petermaxlee · 2016-08-20T06:40:32Z

+/**
+ * User specified options for file streams.
+ */
+class FileStreamOptions(@transient private val parameters: Map[String, String])


This is a similar setup to CSVOptions and JSONOptions. I felt it would be easier to track the list of options read by the source here.

You can remove Serializable and @transient from this class. It's not used in the executor side.

petermaxlee · 2016-08-20T06:44:36Z

cc @rxin @cloud-fan

rxin · 2016-08-20T06:54:52Z

+
+ /** Maximum age of a file that can be found in this directory, before it is deleted. */
+ val maxFileAgeMs: Long =
+ Utils.timeStringAsMs(parameters.getOrElse("maxFileAge", "24h"))


24 hour seems too short. Maybe a month or a week?

rxin · 2016-08-20T07:25:08Z

cc @tdas

SparkQA · 2016-08-20T08:39:44Z

Test build #64133 has finished for PR 14728 at commit ce1dd9c.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- case class FileEntry(path: String, timestamp: Timestamp) extends Serializable
- class SeenFilesMap(maxAgeMs: Long)

marmbrus · 2016-08-23T00:51:12Z

 /**
- * A very simple source that reads text files from the given directory as they appear.
- *
- * TODO Clean up the metadata files periodically


This TODO still applies right?

Put it back. Also updated the file with a new test case to make the seen map more robust.

SparkQA · 2016-08-23T04:58:10Z

Test build #64260 has finished for PR 14728 at commit a371f05.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

petermaxlee · 2016-08-24T06:43:44Z

I've updated the default and set it to 1 week.

rxin · 2016-08-24T06:50:13Z

@jerryshao want to review this?

jerryshao · 2016-08-24T06:55:18Z

Sure, let me take a look at this.

jerryshao · 2016-08-24T08:07:41Z

+
+ def apply(paramName: String, paramValue: String): FileStreamOptions = {
+ new FileStreamOptions(Map(paramName -> paramValue))
+ }


Looks like these two apply methods are not used, it would be better to remove them if not used.

SparkQA · 2016-08-24T08:41:14Z

Test build #64336 has finished for PR 14728 at commit 6817dec.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

jerryshao · 2016-08-24T08:50:02Z

For the definition of maxAge, currently from the code it is max age to latest file, people may misunderstand it is max age to current time, so it would be better to document the meaning.

Another concern is that this patch changes the MetadataLog format, which will break upgrade compatibility, also user with old metadata log existed will encounter exception when using new code.

petermaxlee · 2016-08-24T17:00:58Z

Re: format compatibility

My understanding is that structured streaming is not yet production ready and the compatibility guarantee across versions doesn't really apply. Otherwise, we shouldn't use Java serialization here at all and should opt for explicit JSON or protobuf.

SparkQA · 2016-08-24T19:01:29Z

Test build #64362 has finished for PR 14728 at commit 04a112d.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

zsxwing · 2016-08-26T00:19:17Z

Looks pretty good. Just one comment about Serializable.

SparkQA · 2016-08-26T06:15:22Z

Test build #64454 has finished for PR 14728 at commit 9a5ed19.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- class FileStreamOptions(parameters: Map[String, String]) extends Logging

zsxwing · 2016-08-26T18:29:40Z

LGTM. Thanks! Merging into master and 2.0.

… files indefinitely ## What changes were proposed in this pull request? Before this change, FileStreamSource uses an in-memory hash set to track the list of files processed by the engine. The list can grow indefinitely, leading to OOM or overflow of the hash set. This patch introduces a new user-defined option called "maxFileAge", default to 24 hours. If a file is older than this age, FileStreamSource will purge it from the in-memory map that was used to track the list of files that have been processed. ## How was this patch tested? Added unit tests for the underlying utility, and also added an end-to-end test to validate the purge in FileStreamSourceSuite. Also verified the new test cases would fail when the timeout was set to a very large number. Author: petermaxlee <petermaxlee@gmail.com> Closes#14728 from petermaxlee/SPARK-17165. (cherry picked from commit 9812f7d) Signed-off-by: Shixiong Zhu <shixiong@databricks.com>

[SPARK-17165][SQL] FileStreamSource should not track the list of seen…
ce1dd9c
… files indefinitely

petermaxlee reviewed Aug 20, 2016
View reviewed changes

rxin reviewed Aug 20, 2016
View reviewed changes

marmbrus reviewed Aug 23, 2016
View reviewed changes

More conservative check against lastPurgeTimestamp
a371f05

Set the default max age to 1 week.
6817dec

jerryshao reviewed Aug 24, 2016
View reviewed changes

Code review
04a112d

Remove serializable
9a5ed19

asfgit closed this in 9812f7dAug 26, 2016

frreiss mentioned this pull request Aug 26, 2016
[SPARK-16963] [STREAMING] [SQL] Changes to Source trait and related implementation classes #14553
Closed

frreiss mentioned this pull request Sep 13, 2016
[SPARK-15698][SQL][Streaming] Add the ability to remove the old MetadataLog in FileStreamSource #13513
Closed

Uh oh!

Conversation

petermaxlee commented Aug 20, 2016• edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What changes were proposed in this pull request?

How was this patch tested?

Uh oh!

petermaxleeAug 20, 2016

Choose a reason for hiding this comment

Uh oh!

zsxwingAug 25, 2016

Choose a reason for hiding this comment

Uh oh!

petermaxlee commented Aug 20, 2016

Uh oh!

rxinAug 20, 2016• edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

rxin commented Aug 20, 2016

Uh oh!

SparkQA commented Aug 20, 2016

Uh oh!

marmbrusAug 23, 2016

Choose a reason for hiding this comment

Uh oh!

petermaxleeAug 23, 2016

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Aug 23, 2016

Uh oh!

petermaxlee commented Aug 24, 2016

Uh oh!

rxin commented Aug 24, 2016

Uh oh!

jerryshao commented Aug 24, 2016

Uh oh!

jerryshaoAug 24, 2016

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Aug 24, 2016

Uh oh!

jerryshao commented Aug 24, 2016

Uh oh!

petermaxlee commented Aug 24, 2016

Uh oh!

SparkQA commented Aug 24, 2016

Uh oh!

zsxwing commented Aug 26, 2016

Uh oh!

SparkQA commented Aug 26, 2016

Uh oh!

zsxwing commented Aug 26, 2016

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

7 participants

petermaxlee commented Aug 20, 2016•
edited
Loading

rxinAug 20, 2016•
edited
Loading