[GitHub] [spark] HeartSaVioR commented on a change in pull request #22952: [SPARK-20568][SS] Provide option to clean up completed files in streaming query
HeartSaVioR commented on a change in pull request #22952: [SPARK-20568][SS]
Provide option to clean up completed files in streaming query
URL: https://github.com/apache/spark/pull/22952#discussion_r358624511
##
File path:
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSource.scala
##
@@ -330,4 +341,96 @@ object FileStreamSource {
def size: Int = map.size()
}
+
+ private[sql] trait FileStreamSourceCleaner {
+def clean(entry: FileEntry): Unit
+ }
+
+ private[sql] object FileStreamSourceCleaner {
+def apply(
+fileSystem: FileSystem,
+sourcePath: Path,
+option: FileStreamOptions,
+hadoopConf: Configuration): Option[FileStreamSourceCleaner] =
option.cleanSource match {
+ case CleanSourceMode.ARCHIVE =>
+require(option.sourceArchiveDir.isDefined)
+val path = new Path(option.sourceArchiveDir.get)
+val archiveFs = path.getFileSystem(hadoopConf)
+val qualifiedArchivePath = archiveFs.makeQualified(path)
+Some(new SourceFileArchiver(fileSystem, sourcePath, archiveFs,
qualifiedArchivePath))
+
+ case CleanSourceMode.DELETE =>
+Some(new SourceFileRemover(fileSystem))
+
+ case _ => None
+}
+ }
+
+ private[sql] class SourceFileArchiver(
+ fileSystem: FileSystem,
+ sourcePath: Path,
+ baseArchiveFileSystem: FileSystem,
+ baseArchivePath: Path) extends FileStreamSourceCleaner with Logging {
+assertParameters()
+
+private def assertParameters(): Unit = {
+ require(fileSystem.getUri == baseArchiveFileSystem.getUri, "Base archive
path is located " +
+s"on a different file system than the source files. source path:
$sourcePath" +
+s" / base archive path: $baseArchivePath")
+
+ /**
+ * FileStreamSource reads the files which one of below conditions is met:
+ * 1) file itself is matched with source path
+ * 2) parent directory is matched with source path
Review comment:
FYI, just filed https://issues.apache.org/jira/browse/SPARK-30281 and raised
a patch with picking the option 2. #26845
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on a change in pull request #22952: [SPARK-20568][SS] Provide option to clean up completed files in streaming query
HeartSaVioR commented on a change in pull request #22952: [SPARK-20568][SS]
Provide option to clean up completed files in streaming query
URL: https://github.com/apache/spark/pull/22952#discussion_r358624511
##
File path:
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSource.scala
##
@@ -330,4 +341,96 @@ object FileStreamSource {
def size: Int = map.size()
}
+
+ private[sql] trait FileStreamSourceCleaner {
+def clean(entry: FileEntry): Unit
+ }
+
+ private[sql] object FileStreamSourceCleaner {
+def apply(
+fileSystem: FileSystem,
+sourcePath: Path,
+option: FileStreamOptions,
+hadoopConf: Configuration): Option[FileStreamSourceCleaner] =
option.cleanSource match {
+ case CleanSourceMode.ARCHIVE =>
+require(option.sourceArchiveDir.isDefined)
+val path = new Path(option.sourceArchiveDir.get)
+val archiveFs = path.getFileSystem(hadoopConf)
+val qualifiedArchivePath = archiveFs.makeQualified(path)
+Some(new SourceFileArchiver(fileSystem, sourcePath, archiveFs,
qualifiedArchivePath))
+
+ case CleanSourceMode.DELETE =>
+Some(new SourceFileRemover(fileSystem))
+
+ case _ => None
+}
+ }
+
+ private[sql] class SourceFileArchiver(
+ fileSystem: FileSystem,
+ sourcePath: Path,
+ baseArchiveFileSystem: FileSystem,
+ baseArchivePath: Path) extends FileStreamSourceCleaner with Logging {
+assertParameters()
+
+private def assertParameters(): Unit = {
+ require(fileSystem.getUri == baseArchiveFileSystem.getUri, "Base archive
path is located " +
+s"on a different file system than the source files. source path:
$sourcePath" +
+s" / base archive path: $baseArchivePath")
+
+ /**
+ * FileStreamSource reads the files which one of below conditions is met:
+ * 1) file itself is matched with source path
+ * 2) parent directory is matched with source path
Review comment:
FYI, just filed https://issues.apache.org/jira/browse/SPARK-30281 and raised
a patch with picking the option 2. #26920
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on a change in pull request #22952: [SPARK-20568][SS] Provide option to clean up completed files in streaming query
HeartSaVioR commented on a change in pull request #22952: [SPARK-20568][SS]
Provide option to clean up completed files in streaming query
URL: https://github.com/apache/spark/pull/22952#discussion_r334714556
##
File path:
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSource.scala
##
@@ -258,16 +264,33 @@ class FileStreamSource(
* equal to `end` and will only request offsets greater than `end` in the
future.
*/
override def commit(end: Offset): Unit = {
-// No-op for now; FileStreamSource currently garbage-collects files based
on timestamp
-// and the value of the maxFileAge parameter.
+val logOffset = FileStreamSourceOffset(end).logOffset
+
+if (sourceOptions.cleanSource != CleanSourceMode.NO_OP) {
Review comment:
That was suggested earlier and we agreed to deal with it from follow-up
issue - the PR just has been sitting longer than expected. Could we deal with
follow-up issue?
Btw, if we don't guarantee the cleanup and do it with best effort (actually
we already do it), it won't matter much to do it with background thread -
except the problem you've mentioned. That could be remedied via retaining max
amount of paths to cleanup (yes, this is based on "best effort").
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on a change in pull request #22952: [SPARK-20568][SS] Provide option to clean up completed files in streaming query
HeartSaVioR commented on a change in pull request #22952: [SPARK-20568][SS]
Provide option to clean up completed files in streaming query
URL: https://github.com/apache/spark/pull/22952#discussion_r334718819
##
File path:
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSource.scala
##
@@ -330,4 +353,129 @@ object FileStreamSource {
def size: Int = map.size()
}
+
+ private[sql] class FileStreamSourceCleaner(
+ fileSystem: FileSystem,
+ sourcePath: Path,
+ baseArchivePathString: Option[String]) extends Logging {
+
+private val sourceGlobFilters: Seq[GlobFilter] =
buildSourceGlobFilters(sourcePath)
+
+private val baseArchivePath: Option[Path] = baseArchivePathString.map(new
Path(_))
+
+def archive(entry: FileEntry): Unit = {
+ require(baseArchivePath.isDefined)
+
+ val curPath = new Path(new URI(entry.path))
+ val curPathUri = curPath.toUri
+
+ val newPath = buildArchiveFilePath(curPathUri)
+
+ if (isArchiveFileMatchedAgainstSourcePattern(newPath)) {
+logWarning(s"Fail to move $curPath to $newPath - destination matches "
+
+ s"to source path/pattern. Skip moving file.")
+ } else {
+doArchive(curPath, newPath)
+ }
+}
+
+def delete(entry: FileEntry): Unit = {
+ val curPath = new Path(new URI(entry.path))
+ try {
+logDebug(s"Removing completed file $curPath")
+
+if (!fileSystem.delete(curPath, false)) {
+ logWarning(s"Fail to remove $curPath / skip removing file.")
+}
+ } catch {
+case NonFatal(e) =>
+ // Log to error but swallow exception to avoid process being stopped
+ logWarning(s"Fail to remove $curPath / skip removing file.", e)
+ }
+}
+
+private def buildSourceGlobFilters(sourcePath: Path): Seq[GlobFilter] = {
+ val filters = new scala.collection.mutable.MutableList[GlobFilter]()
+
+ var currentPath = sourcePath
+ while (!currentPath.isRoot) {
+filters += new GlobFilter(currentPath.getName)
+currentPath = currentPath.getParent
+ }
+
+ filters.toList
+}
+
+private def buildArchiveFilePath(pathUri: URI): Path = {
+ require(baseArchivePathString.isDefined)
+ val baseArchivePathStr = baseArchivePathString.get
+ val normalizedBaseArchiveDirPath = if (baseArchivePathStr.endsWith("/"))
{
+baseArchivePathStr.substring(0, baseArchivePathStr.length - 1)
+ } else {
+baseArchivePathStr
+ }
+
+ new Path(normalizedBaseArchiveDirPath + pathUri.getPath)
+}
+
+private def isArchiveFileMatchedAgainstSourcePattern(archiveFile: Path):
Boolean = {
Review comment:
> My guess is that it's checking whether the archive dir is under the source
directory?
It checks whether the destination of archive "file" will be under the source
path (which contains glob) to prevent the possibility of overwriting/re-reading
as input.
So actually the method name contains everything - if `SourcePattern` isn't a
familiar representation, I'll change it to `SourcePath`. If we prefer to
`MatchedWith` instead of `MatchedAgainst`, I'll change it.
If we agree the method name explains what it is doing, we might be able to
skip adding doc - especially it's a private method.
> This looks way too complicated for a simple check like that.
The method becomes complicated because of two reasons:
1) There's a tricky part in FileStreamSource: FileStreamSource doesn't only
match the files which match the source path, but also matches the files which
parent directory matches the source path. So we should consider both cases: 1)
file itself is matched 2) parent directory is matched.
Please refer below comments in test code:
https://github.com/apache/spark/blob/bd8da3799dd160771ebb3ea55b7678b644248425/sql/core/src/test/scala/org/apache/spark/sql/streaming/FileStreamSourceSuite.scala#L736-L778
2) Checking with glob pattern is costly, so we would like to avoid the case
via leveraging known information if possible. For example, when file is moved
to archive directory, destination path will retain input file's path as suffix,
so destination path can't be matched with source path if archive directory's
depth is longer than 2. (Neither file nor parent directory of destination path
can be matched with source path.)
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark
[GitHub] [spark] HeartSaVioR commented on a change in pull request #22952: [SPARK-20568][SS] Provide option to clean up completed files in streaming query
HeartSaVioR commented on a change in pull request #22952: [SPARK-20568][SS]
Provide option to clean up completed files in streaming query
URL: https://github.com/apache/spark/pull/22952#discussion_r334718819
##
File path:
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSource.scala
##
@@ -330,4 +353,129 @@ object FileStreamSource {
def size: Int = map.size()
}
+
+ private[sql] class FileStreamSourceCleaner(
+ fileSystem: FileSystem,
+ sourcePath: Path,
+ baseArchivePathString: Option[String]) extends Logging {
+
+private val sourceGlobFilters: Seq[GlobFilter] =
buildSourceGlobFilters(sourcePath)
+
+private val baseArchivePath: Option[Path] = baseArchivePathString.map(new
Path(_))
+
+def archive(entry: FileEntry): Unit = {
+ require(baseArchivePath.isDefined)
+
+ val curPath = new Path(new URI(entry.path))
+ val curPathUri = curPath.toUri
+
+ val newPath = buildArchiveFilePath(curPathUri)
+
+ if (isArchiveFileMatchedAgainstSourcePattern(newPath)) {
+logWarning(s"Fail to move $curPath to $newPath - destination matches "
+
+ s"to source path/pattern. Skip moving file.")
+ } else {
+doArchive(curPath, newPath)
+ }
+}
+
+def delete(entry: FileEntry): Unit = {
+ val curPath = new Path(new URI(entry.path))
+ try {
+logDebug(s"Removing completed file $curPath")
+
+if (!fileSystem.delete(curPath, false)) {
+ logWarning(s"Fail to remove $curPath / skip removing file.")
+}
+ } catch {
+case NonFatal(e) =>
+ // Log to error but swallow exception to avoid process being stopped
+ logWarning(s"Fail to remove $curPath / skip removing file.", e)
+ }
+}
+
+private def buildSourceGlobFilters(sourcePath: Path): Seq[GlobFilter] = {
+ val filters = new scala.collection.mutable.MutableList[GlobFilter]()
+
+ var currentPath = sourcePath
+ while (!currentPath.isRoot) {
+filters += new GlobFilter(currentPath.getName)
+currentPath = currentPath.getParent
+ }
+
+ filters.toList
+}
+
+private def buildArchiveFilePath(pathUri: URI): Path = {
+ require(baseArchivePathString.isDefined)
+ val baseArchivePathStr = baseArchivePathString.get
+ val normalizedBaseArchiveDirPath = if (baseArchivePathStr.endsWith("/"))
{
+baseArchivePathStr.substring(0, baseArchivePathStr.length - 1)
+ } else {
+baseArchivePathStr
+ }
+
+ new Path(normalizedBaseArchiveDirPath + pathUri.getPath)
+}
+
+private def isArchiveFileMatchedAgainstSourcePattern(archiveFile: Path):
Boolean = {
Review comment:
> My guess is that it's checking whether the archive dir is under the source
directory?
It checks whether the destination of archive "file" will be under the source
path (which contains glob) to prevent the possibility of overwriting/re-reading
as input.
So actually the method name contains everything - if `SourcePattern` isn't a
familiar representation, I'll change it to `SourcePath`. If we prefer to
`MatchedWith` instead of `MatchedAgainst`, I'll change it.
If we agree the method name explains what it is doing, we might be able to
skip adding doc - especially it's a private method. It may be the thing where
is the best place to explain the details - I've added comments near code lines,
but I can adjust it to scaladoc if we prefer it.
> This looks way too complicated for a simple check like that.
The method becomes complicated because of two reasons:
1) There's a tricky part in FileStreamSource: FileStreamSource doesn't only
match the files which match the source path, but also matches the files which
parent directory matches the source path. So we should consider both cases: 1)
file itself is matched 2) parent directory is matched.
Please refer below comments in test code:
https://github.com/apache/spark/blob/bd8da3799dd160771ebb3ea55b7678b644248425/sql/core/src/test/scala/org/apache/spark/sql/streaming/FileStreamSourceSuite.scala#L736-L778
2) Checking with glob pattern is costly, so we would like to avoid the case
via leveraging known information if possible. For example, when file is moved
to archive directory, destination path will retain input file's path as suffix,
so destination path can't be matched with source path if archive directory's
depth is longer than 2. (Neither file nor parent directory of destination path
can be matched with source path.)
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
-
[GitHub] [spark] HeartSaVioR commented on a change in pull request #22952: [SPARK-20568][SS] Provide option to clean up completed files in streaming query
HeartSaVioR commented on a change in pull request #22952: [SPARK-20568][SS]
Provide option to clean up completed files in streaming query
URL: https://github.com/apache/spark/pull/22952#discussion_r334720763
##
File path:
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSource.scala
##
@@ -330,4 +353,129 @@ object FileStreamSource {
def size: Int = map.size()
}
+
+ private[sql] class FileStreamSourceCleaner(
+ fileSystem: FileSystem,
+ sourcePath: Path,
+ baseArchivePathString: Option[String]) extends Logging {
+
+private val sourceGlobFilters: Seq[GlobFilter] =
buildSourceGlobFilters(sourcePath)
+
+private val baseArchivePath: Option[Path] = baseArchivePathString.map(new
Path(_))
+
+def archive(entry: FileEntry): Unit = {
+ require(baseArchivePath.isDefined)
+
+ val curPath = new Path(new URI(entry.path))
+ val curPathUri = curPath.toUri
+
+ val newPath = buildArchiveFilePath(curPathUri)
+
+ if (isArchiveFileMatchedAgainstSourcePattern(newPath)) {
+logWarning(s"Fail to move $curPath to $newPath - destination matches "
+
+ s"to source path/pattern. Skip moving file.")
+ } else {
+doArchive(curPath, newPath)
+ }
+}
+
+def delete(entry: FileEntry): Unit = {
+ val curPath = new Path(new URI(entry.path))
+ try {
+logDebug(s"Removing completed file $curPath")
+
+if (!fileSystem.delete(curPath, false)) {
+ logWarning(s"Fail to remove $curPath / skip removing file.")
+}
+ } catch {
+case NonFatal(e) =>
+ // Log to error but swallow exception to avoid process being stopped
+ logWarning(s"Fail to remove $curPath / skip removing file.", e)
+ }
+}
+
+private def buildSourceGlobFilters(sourcePath: Path): Seq[GlobFilter] = {
+ val filters = new scala.collection.mutable.MutableList[GlobFilter]()
+
+ var currentPath = sourcePath
+ while (!currentPath.isRoot) {
+filters += new GlobFilter(currentPath.getName)
+currentPath = currentPath.getParent
+ }
+
+ filters.toList
+}
+
+private def buildArchiveFilePath(pathUri: URI): Path = {
+ require(baseArchivePathString.isDefined)
+ val baseArchivePathStr = baseArchivePathString.get
+ val normalizedBaseArchiveDirPath = if (baseArchivePathStr.endsWith("/"))
{
+baseArchivePathStr.substring(0, baseArchivePathStr.length - 1)
+ } else {
+baseArchivePathStr
+ }
+
+ new Path(normalizedBaseArchiveDirPath + pathUri.getPath)
Review comment:
> Is pathUri guaranteed to contain an absolute path?
Yes, FileStreamSource qualifies the path when listing files.
> Because then this is basically new Path(parentPath,
pathUri.toString.substring(1)), which also may mean you can clean up a bunch of
this code.
`.getPath` is used because URI could have schema part and then
`pathUri.toString.substring(1)` wouldn't work as expected.
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on a change in pull request #22952: [SPARK-20568][SS] Provide option to clean up completed files in streaming query
HeartSaVioR commented on a change in pull request #22952: [SPARK-20568][SS]
Provide option to clean up completed files in streaming query
URL: https://github.com/apache/spark/pull/22952#discussion_r335217980
##
File path:
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSource.scala
##
@@ -330,4 +353,129 @@ object FileStreamSource {
def size: Int = map.size()
}
+
+ private[sql] class FileStreamSourceCleaner(
+ fileSystem: FileSystem,
+ sourcePath: Path,
+ baseArchivePathString: Option[String]) extends Logging {
+
+private val sourceGlobFilters: Seq[GlobFilter] =
buildSourceGlobFilters(sourcePath)
+
+private val baseArchivePath: Option[Path] = baseArchivePathString.map(new
Path(_))
+
+def archive(entry: FileEntry): Unit = {
+ require(baseArchivePath.isDefined)
+
+ val curPath = new Path(new URI(entry.path))
+ val curPathUri = curPath.toUri
+
+ val newPath = buildArchiveFilePath(curPathUri)
+
+ if (isArchiveFileMatchedAgainstSourcePattern(newPath)) {
+logWarning(s"Fail to move $curPath to $newPath - destination matches "
+
+ s"to source path/pattern. Skip moving file.")
+ } else {
+doArchive(curPath, newPath)
+ }
+}
+
+def delete(entry: FileEntry): Unit = {
+ val curPath = new Path(new URI(entry.path))
+ try {
+logDebug(s"Removing completed file $curPath")
+
+if (!fileSystem.delete(curPath, false)) {
+ logWarning(s"Fail to remove $curPath / skip removing file.")
+}
+ } catch {
+case NonFatal(e) =>
+ // Log to error but swallow exception to avoid process being stopped
+ logWarning(s"Fail to remove $curPath / skip removing file.", e)
+ }
+}
+
+private def buildSourceGlobFilters(sourcePath: Path): Seq[GlobFilter] = {
+ val filters = new scala.collection.mutable.MutableList[GlobFilter]()
+
+ var currentPath = sourcePath
+ while (!currentPath.isRoot) {
+filters += new GlobFilter(currentPath.getName)
+currentPath = currentPath.getParent
+ }
+
+ filters.toList
+}
+
+private def buildArchiveFilePath(pathUri: URI): Path = {
+ require(baseArchivePathString.isDefined)
+ val baseArchivePathStr = baseArchivePathString.get
+ val normalizedBaseArchiveDirPath = if (baseArchivePathStr.endsWith("/"))
{
+baseArchivePathStr.substring(0, baseArchivePathStr.length - 1)
+ } else {
+baseArchivePathStr
+ }
+
+ new Path(normalizedBaseArchiveDirPath + pathUri.getPath)
+}
+
+private def isArchiveFileMatchedAgainstSourcePattern(archiveFile: Path):
Boolean = {
Review comment:
I didn't mean it shouldn't be explained in a scaladoc in method - I meant
the method has comments in near code lines to explain. Sounds like the
preference is adding scaladoc in this case; I'll leave a comment.
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on a change in pull request #22952: [SPARK-20568][SS] Provide option to clean up completed files in streaming query
HeartSaVioR commented on a change in pull request #22952: [SPARK-20568][SS]
Provide option to clean up completed files in streaming query
URL: https://github.com/apache/spark/pull/22952#discussion_r335220635
##
File path:
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSource.scala
##
@@ -330,4 +353,129 @@ object FileStreamSource {
def size: Int = map.size()
}
+
+ private[sql] class FileStreamSourceCleaner(
+ fileSystem: FileSystem,
+ sourcePath: Path,
+ baseArchivePathString: Option[String]) extends Logging {
+
+private val sourceGlobFilters: Seq[GlobFilter] =
buildSourceGlobFilters(sourcePath)
+
+private val baseArchivePath: Option[Path] = baseArchivePathString.map(new
Path(_))
+
+def archive(entry: FileEntry): Unit = {
+ require(baseArchivePath.isDefined)
+
+ val curPath = new Path(new URI(entry.path))
+ val curPathUri = curPath.toUri
+
+ val newPath = buildArchiveFilePath(curPathUri)
+
+ if (isArchiveFileMatchedAgainstSourcePattern(newPath)) {
+logWarning(s"Fail to move $curPath to $newPath - destination matches "
+
+ s"to source path/pattern. Skip moving file.")
+ } else {
+doArchive(curPath, newPath)
+ }
+}
+
+def delete(entry: FileEntry): Unit = {
+ val curPath = new Path(new URI(entry.path))
+ try {
+logDebug(s"Removing completed file $curPath")
+
+if (!fileSystem.delete(curPath, false)) {
+ logWarning(s"Fail to remove $curPath / skip removing file.")
+}
+ } catch {
+case NonFatal(e) =>
+ // Log to error but swallow exception to avoid process being stopped
+ logWarning(s"Fail to remove $curPath / skip removing file.", e)
+ }
+}
+
+private def buildSourceGlobFilters(sourcePath: Path): Seq[GlobFilter] = {
+ val filters = new scala.collection.mutable.MutableList[GlobFilter]()
+
+ var currentPath = sourcePath
+ while (!currentPath.isRoot) {
+filters += new GlobFilter(currentPath.getName)
+currentPath = currentPath.getParent
+ }
+
+ filters.toList
+}
+
+private def buildArchiveFilePath(pathUri: URI): Path = {
+ require(baseArchivePathString.isDefined)
+ val baseArchivePathStr = baseArchivePathString.get
+ val normalizedBaseArchiveDirPath = if (baseArchivePathStr.endsWith("/"))
{
+baseArchivePathStr.substring(0, baseArchivePathStr.length - 1)
+ } else {
+baseArchivePathStr
+ }
+
+ new Path(normalizedBaseArchiveDirPath + pathUri.getPath)
Review comment:
Ah I see. I got the point - making child to be relative (by removing leading
'/') and let constructor resolves it. Didn't indicate the trick works. Nice
suggestion.
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on a change in pull request #22952: [SPARK-20568][SS] Provide option to clean up completed files in streaming query
HeartSaVioR commented on a change in pull request #22952: [SPARK-20568][SS]
Provide option to clean up completed files in streaming query
URL: https://github.com/apache/spark/pull/22952#discussion_r338334551
##
File path:
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSource.scala
##
@@ -330,4 +353,137 @@ object FileStreamSource {
def size: Int = map.size()
}
+
+ private[sql] class FileStreamSourceCleaner(
+ fileSystem: FileSystem,
+ sourcePath: Path,
+ baseArchivePathString: Option[String]) extends Logging {
+
+private val sourceGlobFilters: Seq[GlobFilter] =
buildSourceGlobFilters(sourcePath)
+
+private val baseArchivePath: Option[Path] = baseArchivePathString.map(new
Path(_))
+
+def archive(entry: FileEntry): Unit = {
+ require(baseArchivePath.isDefined)
+
+ val curPath = new Path(new URI(entry.path))
+ val curPathUri = curPath.toUri
+
+ val newPath = buildArchiveFilePath(curPathUri)
+
+ if (isArchiveFileMatchedAgainstSourcePattern(newPath)) {
+logWarning(s"Fail to move $curPath to $newPath - destination matches "
+
+ s"to source path/pattern. Skip moving file.")
+ } else {
+doArchive(curPath, newPath)
+ }
+}
+
+def delete(entry: FileEntry): Unit = {
+ val curPath = new Path(new URI(entry.path))
+ try {
+logDebug(s"Removing completed file $curPath")
+
+if (!fileSystem.delete(curPath, false)) {
+ logWarning(s"Fail to remove $curPath / skip removing file.")
+}
+ } catch {
+case NonFatal(e) =>
+ // Log to error but swallow exception to avoid process being stopped
+ logWarning(s"Fail to remove $curPath / skip removing file.", e)
+ }
+}
+
+private def buildSourceGlobFilters(sourcePath: Path): Seq[GlobFilter] = {
+ val filters = new scala.collection.mutable.MutableList[GlobFilter]()
+
+ var currentPath = sourcePath
+ while (!currentPath.isRoot) {
+filters += new GlobFilter(currentPath.getName)
+currentPath = currentPath.getParent
+ }
+
+ filters.toList
+}
+
+private def buildArchiveFilePath(pathUri: URI): Path = {
+ require(baseArchivePath.isDefined)
+ new Path(baseArchivePath.get, pathUri.getPath.stripPrefix("/"))
+}
+
+/**
+ * This method checks whether the destination of archive file will be
under the source path
+ * (which contains glob) to prevent the possibility of
overwriting/re-reading as input.
+ *
+ * FileStreamSource reads the files which one of below conditions is met:
+ * 1) file itself is matched with source path
+ * 2) parent directory is matched with source path
+ *
+ * Checking with glob pattern is costly, so this method leverages above
information to prune
+ * the cases where the file cannot be matched with source path. For
example, when file is
+ * moved to archive directory, destination path will retain input file's
path as suffix,
+ * so destination path can't be matched with source path if archive
directory's depth is
+ * longer than 2, as neither file nor parent directory of destination path
can be matched
+ * with source path.
+ */
+private def isArchiveFileMatchedAgainstSourcePattern(archiveFile: Path):
Boolean = {
+ if (baseArchivePath.get.depth() > 2) {
Review comment:
This optimization deals with simple fact: "glob path would only cover files
in specific depth" (there's no notation like `**` which would match multiple
depths of directories), so the depth of possibly matched paths is deterministic
if you know about glob path.
Given we move source file on the archive path which would be a directory,
depth will be increased by 1. And FileStreamSource allows the "parent
directory" of source file to be matched with the glob path, the possible
matching depths would be "the depth of glob path" + 1 and "the depth of glob
path" + 2. Does it make sense?
> is there a reason why you need to check this for every archived path?
Can't you, during config validation, perform this check once, with some path
generated to match the source pattern, and then declare that there is no
conflict between the archive path and the source path?
No, that's what I've been missing. You're right it can be determined once we
initialize. Nice finding.
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on a change in pull request #22952: [SPARK-20568][SS] Provide option to clean up completed files in streaming query
HeartSaVioR commented on a change in pull request #22952: [SPARK-20568][SS] Provide option to clean up completed files in streaming query URL: https://github.com/apache/spark/pull/22952#discussion_r338336181 ## File path: docs/structured-streaming-programming-guide.md ## @@ -546,6 +546,13 @@ Here are the details of all the sources in Spark. "s3://a/dataset.txt" "s3n://a/b/dataset.txt" "s3a://a/b/c/dataset.txt" +cleanSource: option to clean up completed files after processing. +Available options are "archive", "delete", "off". If the option is not provided, the default value is "off". +When "archive" is provided, additional option sourceArchiveDir must be provided as well. The value of "sourceArchiveDir" must be outside of source path, to ensure archived files are never included to new source files again. +Spark will move source files respecting its own path. For example, if the path of source file is "/a/b/dataset.txt" and the path of archive directory is "/archived/here", file will be moved to "/archived/here/a/b/dataset.txt" +NOTE: Both archiving (via moving) or deleting completed files would introduce overhead (slow down) in each micro-batch, so you need to understand the cost for each operation in your file system before enabling this option. On the other hand, enabling this option will reduce the cost to list source files which is considered as a heavy operation. +NOTE 2: The source path should not be used from multiple sources or queries when enabling this option, because source files will be moved or deleted which behavior may impact the other sources and queries. Review comment: This is a reflection of previous review comment https://github.com/apache/spark/pull/22952#discussion_r237314459 This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services - To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on a change in pull request #22952: [SPARK-20568][SS] Provide option to clean up completed files in streaming query
HeartSaVioR commented on a change in pull request #22952: [SPARK-20568][SS] Provide option to clean up completed files in streaming query URL: https://github.com/apache/spark/pull/22952#discussion_r338338952 ## File path: docs/structured-streaming-programming-guide.md ## @@ -546,6 +546,13 @@ Here are the details of all the sources in Spark. "s3://a/dataset.txt" "s3n://a/b/dataset.txt" "s3a://a/b/c/dataset.txt" +cleanSource: option to clean up completed files after processing. +Available options are "archive", "delete", "off". If the option is not provided, the default value is "off". +When "archive" is provided, additional option sourceArchiveDir must be provided as well. The value of "sourceArchiveDir" must be outside of source path, to ensure archived files are never included to new source files again. +Spark will move source files respecting its own path. For example, if the path of source file is "/a/b/dataset.txt" and the path of archive directory is "/archived/here", file will be moved to "/archived/here/a/b/dataset.txt" +NOTE: Both archiving (via moving) or deleting completed files would introduce overhead (slow down) in each micro-batch, so you need to understand the cost for each operation in your file system before enabling this option. On the other hand, enabling this option will reduce the cost to list source files which is considered as a heavy operation. +NOTE 2: The source path should not be used from multiple sources or queries when enabling this option, because source files will be moved or deleted which behavior may impact the other sources and queries. Review comment: Revisiting the comment throughly and yes you're right that was the intention. OK to remove latter part. This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services - To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on a change in pull request #22952: [SPARK-20568][SS] Provide option to clean up completed files in streaming query
HeartSaVioR commented on a change in pull request #22952: [SPARK-20568][SS]
Provide option to clean up completed files in streaming query
URL: https://github.com/apache/spark/pull/22952#discussion_r338347720
##
File path:
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSource.scala
##
@@ -330,4 +353,137 @@ object FileStreamSource {
def size: Int = map.size()
}
+
+ private[sql] class FileStreamSourceCleaner(
+ fileSystem: FileSystem,
+ sourcePath: Path,
+ baseArchivePathString: Option[String]) extends Logging {
+
+private val sourceGlobFilters: Seq[GlobFilter] =
buildSourceGlobFilters(sourcePath)
+
+private val baseArchivePath: Option[Path] = baseArchivePathString.map(new
Path(_))
+
+def archive(entry: FileEntry): Unit = {
+ require(baseArchivePath.isDefined)
+
+ val curPath = new Path(new URI(entry.path))
+ val curPathUri = curPath.toUri
+
+ val newPath = buildArchiveFilePath(curPathUri)
+
+ if (isArchiveFileMatchedAgainstSourcePattern(newPath)) {
+logWarning(s"Fail to move $curPath to $newPath - destination matches "
+
+ s"to source path/pattern. Skip moving file.")
+ } else {
+doArchive(curPath, newPath)
+ }
+}
+
+def delete(entry: FileEntry): Unit = {
+ val curPath = new Path(new URI(entry.path))
+ try {
+logDebug(s"Removing completed file $curPath")
+
+if (!fileSystem.delete(curPath, false)) {
+ logWarning(s"Fail to remove $curPath / skip removing file.")
+}
+ } catch {
+case NonFatal(e) =>
+ // Log to error but swallow exception to avoid process being stopped
+ logWarning(s"Fail to remove $curPath / skip removing file.", e)
+ }
+}
+
+private def buildSourceGlobFilters(sourcePath: Path): Seq[GlobFilter] = {
+ val filters = new scala.collection.mutable.MutableList[GlobFilter]()
+
+ var currentPath = sourcePath
+ while (!currentPath.isRoot) {
+filters += new GlobFilter(currentPath.getName)
+currentPath = currentPath.getParent
+ }
+
+ filters.toList
+}
+
+private def buildArchiveFilePath(pathUri: URI): Path = {
+ require(baseArchivePath.isDefined)
+ new Path(baseArchivePath.get, pathUri.getPath.stripPrefix("/"))
+}
+
+/**
+ * This method checks whether the destination of archive file will be
under the source path
+ * (which contains glob) to prevent the possibility of
overwriting/re-reading as input.
+ *
+ * FileStreamSource reads the files which one of below conditions is met:
+ * 1) file itself is matched with source path
+ * 2) parent directory is matched with source path
+ *
+ * Checking with glob pattern is costly, so this method leverages above
information to prune
+ * the cases where the file cannot be matched with source path. For
example, when file is
+ * moved to archive directory, destination path will retain input file's
path as suffix,
+ * so destination path can't be matched with source path if archive
directory's depth is
+ * longer than 2, as neither file nor parent directory of destination path
can be matched
+ * with source path.
+ */
+private def isArchiveFileMatchedAgainstSourcePattern(archiveFile: Path):
Boolean = {
+ if (baseArchivePath.get.depth() > 2) {
+// there's no chance for archive file to be matched against source
pattern
+return false
+ }
+
+ var matched = true
+
+ // new path will never match against source path when the depth is not a
range of
+ // the depth of source path ~ (the depth of source path + 1)
+ // because the source files are picked when they match against source
pattern or
+ // their parent directories match against source pattern
+ val depthSourcePattern = sourceGlobFilters.length
+ val depthArchiveFile = archiveFile.depth()
+
+ // we already checked against the depth of archive path, but rechecking
wouldn't hurt
+ if (depthArchiveFile < depthSourcePattern || depthArchiveFile >
depthSourcePattern + 1) {
+// never matched
+matched = false
+ } else {
+var pathToCompare = if (depthArchiveFile == depthSourcePattern + 1) {
+ archiveFile.getParent
+} else {
+ archiveFile
+}
+
+// Now pathToCompare should have same depth as sourceGlobFilters.length
+var index = 0
+do {
+ // GlobFilter only matches against its name, not full path so it's
safe to compare
+ if (!sourceGlobFilters(index).accept(pathToCompare)) {
+matched = false
+ } else {
+pathToCompare = pathToCompare.getParent
+index += 1
+ }
+} while (matched && !pathToCompare.isRoot)
+ }
+
+ matched
+}
+
+private def doArchive(sourcePath: Path, archivePath: Path): Unit = {
+
[GitHub] [spark] HeartSaVioR commented on a change in pull request #22952: [SPARK-20568][SS] Provide option to clean up completed files in streaming query
HeartSaVioR commented on a change in pull request #22952: [SPARK-20568][SS]
Provide option to clean up completed files in streaming query
URL: https://github.com/apache/spark/pull/22952#discussion_r338334551
##
File path:
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSource.scala
##
@@ -330,4 +353,137 @@ object FileStreamSource {
def size: Int = map.size()
}
+
+ private[sql] class FileStreamSourceCleaner(
+ fileSystem: FileSystem,
+ sourcePath: Path,
+ baseArchivePathString: Option[String]) extends Logging {
+
+private val sourceGlobFilters: Seq[GlobFilter] =
buildSourceGlobFilters(sourcePath)
+
+private val baseArchivePath: Option[Path] = baseArchivePathString.map(new
Path(_))
+
+def archive(entry: FileEntry): Unit = {
+ require(baseArchivePath.isDefined)
+
+ val curPath = new Path(new URI(entry.path))
+ val curPathUri = curPath.toUri
+
+ val newPath = buildArchiveFilePath(curPathUri)
+
+ if (isArchiveFileMatchedAgainstSourcePattern(newPath)) {
+logWarning(s"Fail to move $curPath to $newPath - destination matches "
+
+ s"to source path/pattern. Skip moving file.")
+ } else {
+doArchive(curPath, newPath)
+ }
+}
+
+def delete(entry: FileEntry): Unit = {
+ val curPath = new Path(new URI(entry.path))
+ try {
+logDebug(s"Removing completed file $curPath")
+
+if (!fileSystem.delete(curPath, false)) {
+ logWarning(s"Fail to remove $curPath / skip removing file.")
+}
+ } catch {
+case NonFatal(e) =>
+ // Log to error but swallow exception to avoid process being stopped
+ logWarning(s"Fail to remove $curPath / skip removing file.", e)
+ }
+}
+
+private def buildSourceGlobFilters(sourcePath: Path): Seq[GlobFilter] = {
+ val filters = new scala.collection.mutable.MutableList[GlobFilter]()
+
+ var currentPath = sourcePath
+ while (!currentPath.isRoot) {
+filters += new GlobFilter(currentPath.getName)
+currentPath = currentPath.getParent
+ }
+
+ filters.toList
+}
+
+private def buildArchiveFilePath(pathUri: URI): Path = {
+ require(baseArchivePath.isDefined)
+ new Path(baseArchivePath.get, pathUri.getPath.stripPrefix("/"))
+}
+
+/**
+ * This method checks whether the destination of archive file will be
under the source path
+ * (which contains glob) to prevent the possibility of
overwriting/re-reading as input.
+ *
+ * FileStreamSource reads the files which one of below conditions is met:
+ * 1) file itself is matched with source path
+ * 2) parent directory is matched with source path
+ *
+ * Checking with glob pattern is costly, so this method leverages above
information to prune
+ * the cases where the file cannot be matched with source path. For
example, when file is
+ * moved to archive directory, destination path will retain input file's
path as suffix,
+ * so destination path can't be matched with source path if archive
directory's depth is
+ * longer than 2, as neither file nor parent directory of destination path
can be matched
+ * with source path.
+ */
+private def isArchiveFileMatchedAgainstSourcePattern(archiveFile: Path):
Boolean = {
+ if (baseArchivePath.get.depth() > 2) {
Review comment:
This optimization deals with simple fact: "glob path would only cover files
in specific depth" (there's no notation like `**` which would match multiple
depths of directories), so the depth of possibly matched paths is deterministic
if you know about glob path.
Suppose you have combined the path, all possible cases of archive path to be
matched with glob path are 1) direct match: only possible if base archive path
is just '/' which is depth 1 2) parent dir match: direct subdirectory under /
(like `/a`) which the depth of base archive path would be 2. Does it make sense?
> is there a reason why you need to check this for every archived path?
Can't you, during config validation, perform this check once, with some path
generated to match the source pattern, and then declare that there is no
conflict between the archive path and the source path?
No, that's what I've been missing. You're right it can be determined once we
initialize. Nice finding.
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on a change in pull request #22952: [SPARK-20568][SS] Provide option to clean up completed files in streaming query
HeartSaVioR commented on a change in pull request #22952: [SPARK-20568][SS]
Provide option to clean up completed files in streaming query
URL: https://github.com/apache/spark/pull/22952#discussion_r338835194
##
File path:
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSource.scala
##
@@ -330,4 +362,139 @@ object FileStreamSource {
def size: Int = map.size()
}
+
+ private[sql] class FileStreamSourceCleaner(
+ fileSystem: FileSystem,
+ sourcePath: Path,
+ baseArchiveFileSystem: Option[FileSystem],
+ baseArchivePath: Option[Path]) extends Logging {
+require(baseArchiveFileSystem.isDefined == baseArchivePath.isDefined)
+
+private val sourceGlobFilters: Seq[GlobFilter] =
buildSourceGlobFilters(sourcePath)
+
+private val sameFsSourceAndArchive: Boolean = {
Review comment:
Oh OK. Sounds OK to fail the query and it might be even better than
realizing later the configuration doesn't work since it was misconfigured or
not supported. I'll make the change.
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on a change in pull request #22952: [SPARK-20568][SS] Provide option to clean up completed files in streaming query
HeartSaVioR commented on a change in pull request #22952: [SPARK-20568][SS]
Provide option to clean up completed files in streaming query
URL: https://github.com/apache/spark/pull/22952#discussion_r338837229
##
File path:
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSource.scala
##
@@ -330,4 +362,139 @@ object FileStreamSource {
def size: Int = map.size()
}
+
+ private[sql] class FileStreamSourceCleaner(
+ fileSystem: FileSystem,
+ sourcePath: Path,
+ baseArchiveFileSystem: Option[FileSystem],
+ baseArchivePath: Option[Path]) extends Logging {
+require(baseArchiveFileSystem.isDefined == baseArchivePath.isDefined)
+
+private val sourceGlobFilters: Seq[GlobFilter] =
buildSourceGlobFilters(sourcePath)
+
+private val sameFsSourceAndArchive: Boolean = {
+ baseArchiveFileSystem.exists { fs =>
+if (fileSystem.getUri != fs.getUri) {
+ logWarning("Base archive path is located to the different filesystem
with source, " +
+s"which is not supported. source path: ${sourcePath} / base
archive path: " +
+s"${baseArchivePath.get}")
+ false
+} else {
+ true
+}
+ }
+}
+
+/**
+ * This is a flag to skip matching archived path with source path.
+ *
+ * FileStreamSource reads the files which one of below conditions is met:
+ * 1) file itself is matched with source path
+ * 2) parent directory is matched with source path
+ *
+ * Checking with glob pattern is costly, so this flag leverages above
information to prune
+ * the cases where the file cannot be matched with source path. For
example, when file is
+ * moved to archive directory, destination path will retain input file's
path as suffix,
+ * so destination path can't be matched with source path if archive
directory's depth is
+ * longer than 2, as neither file nor parent directory of destination path
can be matched
+ * with source path.
+ */
+private val skipCheckingGlob: Boolean = baseArchivePath.exists(_.depth() >
2)
+
+def archive(entry: FileEntry): Unit = {
+ require(baseArchivePath.isDefined)
+
+ if (sameFsSourceAndArchive) {
+val curPath = new Path(new URI(entry.path))
+val newPath = new Path(baseArchivePath.get,
curPath.toUri.getPath.stripPrefix("/"))
+
+if (!skipCheckingGlob && pathMatchesSourcePattern(newPath)) {
Review comment:
Not sure I understand. Could you elaborate? I agreed with your suggestion as
it shouldn't be needed to be calculated per path and don't get anything except
that.
If you meant about moving skipCheckingGlob into config validation, I'm not
sure I like it, as it's just the optimization on FileStreamSourceCleaner,
implementation details, so no need to expose it to outside.
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on a change in pull request #22952: [SPARK-20568][SS] Provide option to clean up completed files in streaming query
HeartSaVioR commented on a change in pull request #22952: [SPARK-20568][SS]
Provide option to clean up completed files in streaming query
URL: https://github.com/apache/spark/pull/22952#discussion_r338837940
##
File path:
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSource.scala
##
@@ -330,4 +362,139 @@ object FileStreamSource {
def size: Int = map.size()
}
+
+ private[sql] class FileStreamSourceCleaner(
+ fileSystem: FileSystem,
+ sourcePath: Path,
+ baseArchiveFileSystem: Option[FileSystem],
+ baseArchivePath: Option[Path]) extends Logging {
+require(baseArchiveFileSystem.isDefined == baseArchivePath.isDefined)
+
+private val sourceGlobFilters: Seq[GlobFilter] =
buildSourceGlobFilters(sourcePath)
+
+private val sameFsSourceAndArchive: Boolean = {
+ baseArchiveFileSystem.exists { fs =>
+if (fileSystem.getUri != fs.getUri) {
+ logWarning("Base archive path is located to the different filesystem
with source, " +
+s"which is not supported. source path: ${sourcePath} / base
archive path: " +
+s"${baseArchivePath.get}")
+ false
+} else {
+ true
+}
+ }
+}
+
+/**
+ * This is a flag to skip matching archived path with source path.
+ *
+ * FileStreamSource reads the files which one of below conditions is met:
+ * 1) file itself is matched with source path
+ * 2) parent directory is matched with source path
+ *
+ * Checking with glob pattern is costly, so this flag leverages above
information to prune
+ * the cases where the file cannot be matched with source path. For
example, when file is
+ * moved to archive directory, destination path will retain input file's
path as suffix,
+ * so destination path can't be matched with source path if archive
directory's depth is
+ * longer than 2, as neither file nor parent directory of destination path
can be matched
+ * with source path.
+ */
+private val skipCheckingGlob: Boolean = baseArchivePath.exists(_.depth() >
2)
+
+def archive(entry: FileEntry): Unit = {
+ require(baseArchivePath.isDefined)
+
+ if (sameFsSourceAndArchive) {
+val curPath = new Path(new URI(entry.path))
+val newPath = new Path(baseArchivePath.get,
curPath.toUri.getPath.stripPrefix("/"))
+
+if (!skipCheckingGlob && pathMatchesSourcePattern(newPath)) {
Review comment:
OK looking the comment again I only understood small amount of your comment.
Sorry about this. The source path contains wildcard in any subpath so only
depth check can be optimized. Please correct me if I'm missing here.
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on a change in pull request #22952: [SPARK-20568][SS] Provide option to clean up completed files in streaming query
HeartSaVioR commented on a change in pull request #22952: [SPARK-20568][SS]
Provide option to clean up completed files in streaming query
URL: https://github.com/apache/spark/pull/22952#discussion_r338837940
##
File path:
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSource.scala
##
@@ -330,4 +362,139 @@ object FileStreamSource {
def size: Int = map.size()
}
+
+ private[sql] class FileStreamSourceCleaner(
+ fileSystem: FileSystem,
+ sourcePath: Path,
+ baseArchiveFileSystem: Option[FileSystem],
+ baseArchivePath: Option[Path]) extends Logging {
+require(baseArchiveFileSystem.isDefined == baseArchivePath.isDefined)
+
+private val sourceGlobFilters: Seq[GlobFilter] =
buildSourceGlobFilters(sourcePath)
+
+private val sameFsSourceAndArchive: Boolean = {
+ baseArchiveFileSystem.exists { fs =>
+if (fileSystem.getUri != fs.getUri) {
+ logWarning("Base archive path is located to the different filesystem
with source, " +
+s"which is not supported. source path: ${sourcePath} / base
archive path: " +
+s"${baseArchivePath.get}")
+ false
+} else {
+ true
+}
+ }
+}
+
+/**
+ * This is a flag to skip matching archived path with source path.
+ *
+ * FileStreamSource reads the files which one of below conditions is met:
+ * 1) file itself is matched with source path
+ * 2) parent directory is matched with source path
+ *
+ * Checking with glob pattern is costly, so this flag leverages above
information to prune
+ * the cases where the file cannot be matched with source path. For
example, when file is
+ * moved to archive directory, destination path will retain input file's
path as suffix,
+ * so destination path can't be matched with source path if archive
directory's depth is
+ * longer than 2, as neither file nor parent directory of destination path
can be matched
+ * with source path.
+ */
+private val skipCheckingGlob: Boolean = baseArchivePath.exists(_.depth() >
2)
+
+def archive(entry: FileEntry): Unit = {
+ require(baseArchivePath.isDefined)
+
+ if (sameFsSourceAndArchive) {
+val curPath = new Path(new URI(entry.path))
+val newPath = new Path(baseArchivePath.get,
curPath.toUri.getPath.stripPrefix("/"))
+
+if (!skipCheckingGlob && pathMatchesSourcePattern(newPath)) {
Review comment:
OK looking the comment again I only understood small amount of your comment.
Sorry about this. ~~The source path contains wildcard in any subpath so only
depth check can be optimized. Please correct me if I'm missing here.~~
Sorry never mind. I just got your point about further pruning the case. If
baseArchivePath doesn't match with source path, no need to check further. Will
add.
Btw, I guess we still need this if baseArchivePath matches with source path,
but please let me know if you're seeing the approach of verification we can
eliminate this entirely.
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on a change in pull request #22952: [SPARK-20568][SS] Provide option to clean up completed files in streaming query
HeartSaVioR commented on a change in pull request #22952: [SPARK-20568][SS]
Provide option to clean up completed files in streaming query
URL: https://github.com/apache/spark/pull/22952#discussion_r338847897
##
File path:
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSource.scala
##
@@ -330,4 +362,139 @@ object FileStreamSource {
def size: Int = map.size()
}
+
+ private[sql] class FileStreamSourceCleaner(
+ fileSystem: FileSystem,
+ sourcePath: Path,
+ baseArchiveFileSystem: Option[FileSystem],
+ baseArchivePath: Option[Path]) extends Logging {
+require(baseArchiveFileSystem.isDefined == baseArchivePath.isDefined)
+
+private val sourceGlobFilters: Seq[GlobFilter] =
buildSourceGlobFilters(sourcePath)
+
+private val sameFsSourceAndArchive: Boolean = {
+ baseArchiveFileSystem.exists { fs =>
+if (fileSystem.getUri != fs.getUri) {
+ logWarning("Base archive path is located to the different filesystem
with source, " +
+s"which is not supported. source path: ${sourcePath} / base
archive path: " +
+s"${baseArchivePath.get}")
+ false
+} else {
+ true
+}
+ }
+}
+
+/**
+ * This is a flag to skip matching archived path with source path.
+ *
+ * FileStreamSource reads the files which one of below conditions is met:
+ * 1) file itself is matched with source path
+ * 2) parent directory is matched with source path
+ *
+ * Checking with glob pattern is costly, so this flag leverages above
information to prune
+ * the cases where the file cannot be matched with source path. For
example, when file is
+ * moved to archive directory, destination path will retain input file's
path as suffix,
+ * so destination path can't be matched with source path if archive
directory's depth is
+ * longer than 2, as neither file nor parent directory of destination path
can be matched
+ * with source path.
+ */
+private val skipCheckingGlob: Boolean = baseArchivePath.exists(_.depth() >
2)
+
+def archive(entry: FileEntry): Unit = {
+ require(baseArchivePath.isDefined)
+
+ if (sameFsSourceAndArchive) {
+val curPath = new Path(new URI(entry.path))
+val newPath = new Path(baseArchivePath.get,
curPath.toUri.getPath.stripPrefix("/"))
+
+if (!skipCheckingGlob && pathMatchesSourcePattern(newPath)) {
Review comment:
> I don't fully understand the matching you explained, but my guess is that
you could end up with an archived file called "/hello/hello1/hell" and that
somehow matches the pattern?
Exactly. We cannot forecast if first depth of source path has wildcard. Yes
we can also prune the case if the first depth of source path doesn't contain
wildcard, but feels like more and more complicated.
As we would really want to simplify the case, how about just forcing
baseArchivePath to have depth more than 2? (I feel it would be harmless in
production - if they're encountering the case, just create subdirectory.) Then
we can eliminate the pattern check entirely and throw error on config
validation phase. WDYT?
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on a change in pull request #22952: [SPARK-20568][SS] Provide option to clean up completed files in streaming query
HeartSaVioR commented on a change in pull request #22952: [SPARK-20568][SS]
Provide option to clean up completed files in streaming query
URL: https://github.com/apache/spark/pull/22952#discussion_r338960691
##
File path:
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSource.scala
##
@@ -330,4 +362,139 @@ object FileStreamSource {
def size: Int = map.size()
}
+
+ private[sql] class FileStreamSourceCleaner(
+ fileSystem: FileSystem,
+ sourcePath: Path,
+ baseArchiveFileSystem: Option[FileSystem],
+ baseArchivePath: Option[Path]) extends Logging {
+require(baseArchiveFileSystem.isDefined == baseArchivePath.isDefined)
+
+private val sourceGlobFilters: Seq[GlobFilter] =
buildSourceGlobFilters(sourcePath)
+
+private val sameFsSourceAndArchive: Boolean = {
+ baseArchiveFileSystem.exists { fs =>
+if (fileSystem.getUri != fs.getUri) {
+ logWarning("Base archive path is located to the different filesystem
with source, " +
+s"which is not supported. source path: ${sourcePath} / base
archive path: " +
+s"${baseArchivePath.get}")
+ false
+} else {
+ true
+}
+ }
+}
+
+/**
+ * This is a flag to skip matching archived path with source path.
+ *
+ * FileStreamSource reads the files which one of below conditions is met:
+ * 1) file itself is matched with source path
+ * 2) parent directory is matched with source path
+ *
+ * Checking with glob pattern is costly, so this flag leverages above
information to prune
+ * the cases where the file cannot be matched with source path. For
example, when file is
+ * moved to archive directory, destination path will retain input file's
path as suffix,
+ * so destination path can't be matched with source path if archive
directory's depth is
+ * longer than 2, as neither file nor parent directory of destination path
can be matched
+ * with source path.
+ */
+private val skipCheckingGlob: Boolean = baseArchivePath.exists(_.depth() >
2)
+
+def archive(entry: FileEntry): Unit = {
+ require(baseArchivePath.isDefined)
+
+ if (sameFsSourceAndArchive) {
+val curPath = new Path(new URI(entry.path))
+val newPath = new Path(baseArchivePath.get,
curPath.toUri.getPath.stripPrefix("/"))
+
+if (!skipCheckingGlob && pathMatchesSourcePattern(newPath)) {
Review comment:
Technically, we just need to check pattern only when the depth of base
archive path is 2. The depth of base archive path cannot be 1, because then
archive file will always refer to the same path of original source file. (Yes I
was aware of this but forgot to apply the fact to prune.)
So `depth > 2` -> no need to check, `depth == 1` -> always matches, `depth
== 2` -> need to check the pattern. We can even just simplify the case to avoid
checking per file as checking the earliest dir name of base archive path
matches the earlier dir name (including glob path) of source path. It won't
guarantee all the archive paths will match the source path - that's why I did
full pattern match per file, but if we really don't want to do pattern matching
per file, yes, we can do it.
What I concern about is rather not a technical one. If we want to just yell
and fail the query instead of skipping archive file, we must ensure end users
understand why the query is failed and how to get it fixed. I had to explain
most of reviewers how pattern check works, even with that I'm feeling reviewers
don't feel familiar with it - I'm unsure it would work for end users though we
can add some guidance how it works.
So we will need to simplify the condition as much as possible if we really
want to just fail the query. The difference of necessary amount of explanation
for depth > 2 and depth >=2 are significant, and the actual harm of not
allowing depth == 2 is that they just need to create a subdirectory and use the
subdirectory, which doesn't seem to be critical.
Does it make sense?
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on a change in pull request #22952: [SPARK-20568][SS] Provide option to clean up completed files in streaming query
HeartSaVioR commented on a change in pull request #22952: [SPARK-20568][SS]
Provide option to clean up completed files in streaming query
URL: https://github.com/apache/spark/pull/22952#discussion_r339282089
##
File path:
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSource.scala
##
@@ -330,4 +362,139 @@ object FileStreamSource {
def size: Int = map.size()
}
+
+ private[sql] class FileStreamSourceCleaner(
+ fileSystem: FileSystem,
+ sourcePath: Path,
+ baseArchiveFileSystem: Option[FileSystem],
+ baseArchivePath: Option[Path]) extends Logging {
+require(baseArchiveFileSystem.isDefined == baseArchivePath.isDefined)
+
+private val sourceGlobFilters: Seq[GlobFilter] =
buildSourceGlobFilters(sourcePath)
+
+private val sameFsSourceAndArchive: Boolean = {
+ baseArchiveFileSystem.exists { fs =>
+if (fileSystem.getUri != fs.getUri) {
+ logWarning("Base archive path is located to the different filesystem
with source, " +
+s"which is not supported. source path: ${sourcePath} / base
archive path: " +
+s"${baseArchivePath.get}")
+ false
+} else {
+ true
+}
+ }
+}
+
+/**
+ * This is a flag to skip matching archived path with source path.
+ *
+ * FileStreamSource reads the files which one of below conditions is met:
+ * 1) file itself is matched with source path
+ * 2) parent directory is matched with source path
+ *
+ * Checking with glob pattern is costly, so this flag leverages above
information to prune
+ * the cases where the file cannot be matched with source path. For
example, when file is
+ * moved to archive directory, destination path will retain input file's
path as suffix,
+ * so destination path can't be matched with source path if archive
directory's depth is
+ * longer than 2, as neither file nor parent directory of destination path
can be matched
+ * with source path.
+ */
+private val skipCheckingGlob: Boolean = baseArchivePath.exists(_.depth() >
2)
+
+def archive(entry: FileEntry): Unit = {
+ require(baseArchivePath.isDefined)
+
+ if (sameFsSourceAndArchive) {
+val curPath = new Path(new URI(entry.path))
+val newPath = new Path(baseArchivePath.get,
curPath.toUri.getPath.stripPrefix("/"))
+
+if (!skipCheckingGlob && pathMatchesSourcePattern(newPath)) {
Review comment:
If I were one of end users and query fails because of glob path, I might be
puzzling and want to know why - that's only when we allow `depth == 2` and I
think we've agreed `depth > 2` doesn't need to explain further detail. I'll
apply it.
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on a change in pull request #22952: [SPARK-20568][SS] Provide option to clean up completed files in streaming query
HeartSaVioR commented on a change in pull request #22952: [SPARK-20568][SS]
Provide option to clean up completed files in streaming query
URL: https://github.com/apache/spark/pull/22952#discussion_r340308673
##
File path:
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSource.scala
##
@@ -330,4 +362,77 @@ object FileStreamSource {
def size: Int = map.size()
}
+
+ private[sql] class FileStreamSourceCleaner(
+ fileSystem: FileSystem,
+ sourcePath: Path,
+ baseArchiveFileSystem: Option[FileSystem],
+ baseArchivePath: Option[Path]) extends Logging {
+assertParameters()
+
+private def assertParameters(): Unit = {
+ require(baseArchiveFileSystem.isDefined == baseArchivePath.isDefined)
+
+ baseArchiveFileSystem.foreach { fs =>
+require(fileSystem.getUri == fs.getUri, "Base archive path is located
on a different " +
+ s"file system than the source files. source path: $sourcePath" +
+ s" / base archive path: ${baseArchivePath.get}")
+ }
+
+ baseArchivePath.foreach { path =>
+
+/**
+ * FileStreamSource reads the files which one of below conditions is
met:
+ * 1) file itself is matched with source path
+ * 2) parent directory is matched with source path
+ *
+ * Checking with glob pattern is costly, so set this requirement to
eliminate the cases
+ * where the archive path can be matched with source path. For
example, when file is moved
+ * to archive directory, destination path will retain input file's
path as suffix, so
+ * destination path can't be matched with source path if archive
directory's depth is longer
+ * than 2, as neither file nor parent directory of destination path
can be matched with
+ * source path.
+ */
+require(path.depth() > 2, "Base archive path must have a depth of at
least 2 " +
+ "subdirectories. e.g. '/data/archive'")
+ }
+}
+
+def archive(entry: FileEntry): Unit = {
+ require(baseArchivePath.isDefined)
+
+ val curPath = new Path(new URI(entry.path))
+ val newPath = new Path(baseArchivePath.get,
curPath.toUri.getPath.stripPrefix("/"))
Review comment:
I'm revisiting two issues and not sure there's a viable workaround. Looks
like the issue pointed out was ":" isn't a valid char for HDFS but might be a
valid char for other filesystems so Path API doesn't restrict it and leads
problem. Even HDFS-14762 is closed as "Won't fix".
Would this only occur on `Path(parent, child)` and `Path(pathstr)` is safe?
Would it work if we manually concat two paths as string and pass to Path's
constructor?
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on a change in pull request #22952: [SPARK-20568][SS] Provide option to clean up completed files in streaming query
HeartSaVioR commented on a change in pull request #22952: [SPARK-20568][SS]
Provide option to clean up completed files in streaming query
URL: https://github.com/apache/spark/pull/22952#discussion_r340311674
##
File path:
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSource.scala
##
@@ -330,4 +362,77 @@ object FileStreamSource {
def size: Int = map.size()
}
+
+ private[sql] class FileStreamSourceCleaner(
+ fileSystem: FileSystem,
+ sourcePath: Path,
+ baseArchiveFileSystem: Option[FileSystem],
+ baseArchivePath: Option[Path]) extends Logging {
+assertParameters()
+
+private def assertParameters(): Unit = {
+ require(baseArchiveFileSystem.isDefined == baseArchivePath.isDefined)
+
+ baseArchiveFileSystem.foreach { fs =>
+require(fileSystem.getUri == fs.getUri, "Base archive path is located
on a different " +
+ s"file system than the source files. source path: $sourcePath" +
+ s" / base archive path: ${baseArchivePath.get}")
+ }
+
+ baseArchivePath.foreach { path =>
+
+/**
+ * FileStreamSource reads the files which one of below conditions is
met:
+ * 1) file itself is matched with source path
+ * 2) parent directory is matched with source path
+ *
+ * Checking with glob pattern is costly, so set this requirement to
eliminate the cases
+ * where the archive path can be matched with source path. For
example, when file is moved
+ * to archive directory, destination path will retain input file's
path as suffix, so
+ * destination path can't be matched with source path if archive
directory's depth is longer
+ * than 2, as neither file nor parent directory of destination path
can be matched with
+ * source path.
+ */
+require(path.depth() > 2, "Base archive path must have a depth of at
least 2 " +
+ "subdirectories. e.g. '/data/archive'")
+ }
+}
+
+def archive(entry: FileEntry): Unit = {
+ require(baseArchivePath.isDefined)
+
+ val curPath = new Path(new URI(entry.path))
+ val newPath = new Path(baseArchivePath.get,
curPath.toUri.getPath.stripPrefix("/"))
Review comment:
Thanks for the quick feedback! I'll reflect it.
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on a change in pull request #22952: [SPARK-20568][SS] Provide option to clean up completed files in streaming query
HeartSaVioR commented on a change in pull request #22952: [SPARK-20568][SS]
Provide option to clean up completed files in streaming query
URL: https://github.com/apache/spark/pull/22952#discussion_r341395755
##
File path:
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSource.scala
##
@@ -330,4 +362,77 @@ object FileStreamSource {
def size: Int = map.size()
}
+
+ private[sql] class FileStreamSourceCleaner(
+ fileSystem: FileSystem,
+ sourcePath: Path,
+ baseArchiveFileSystem: Option[FileSystem],
+ baseArchivePath: Option[Path]) extends Logging {
+assertParameters()
+
+private def assertParameters(): Unit = {
+ require(baseArchiveFileSystem.isDefined == baseArchivePath.isDefined)
+
+ baseArchiveFileSystem.foreach { fs =>
+require(fileSystem.getUri == fs.getUri, "Base archive path is located
on a different " +
+ s"file system than the source files. source path: $sourcePath" +
+ s" / base archive path: ${baseArchivePath.get}")
+ }
+
+ baseArchivePath.foreach { path =>
+
+/**
+ * FileStreamSource reads the files which one of below conditions is
met:
+ * 1) file itself is matched with source path
+ * 2) parent directory is matched with source path
+ *
+ * Checking with glob pattern is costly, so set this requirement to
eliminate the cases
+ * where the archive path can be matched with source path. For
example, when file is moved
+ * to archive directory, destination path will retain input file's
path as suffix, so
+ * destination path can't be matched with source path if archive
directory's depth is longer
+ * than 2, as neither file nor parent directory of destination path
can be matched with
+ * source path.
+ */
+require(path.depth() > 2, "Base archive path must have a depth of at
least 2 " +
Review comment:
So the explanation says about 2 "subdirectories", not 2 "depth". / denotes
its own depth. I don't think depth is the term end users are familiar with -
I'll remove the part "a depth of".
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on a change in pull request #22952: [SPARK-20568][SS] Provide option to clean up completed files in streaming query
HeartSaVioR commented on a change in pull request #22952: [SPARK-20568][SS]
Provide option to clean up completed files in streaming query
URL: https://github.com/apache/spark/pull/22952#discussion_r341397990
##
File path:
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSource.scala
##
@@ -53,6 +55,18 @@ class FileStreamSource(
fs.makeQualified(new Path(path)) // can contain glob patterns
}
+ private val sourceCleaner: FileStreamSourceCleaner = {
Review comment:
Yeah, that's a good suggestion. Will address.
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on a change in pull request #22952: [SPARK-20568][SS] Provide option to clean up completed files in streaming query
HeartSaVioR commented on a change in pull request #22952: [SPARK-20568][SS]
Provide option to clean up completed files in streaming query
URL: https://github.com/apache/spark/pull/22952#discussion_r347673657
##
File path:
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSource.scala
##
@@ -330,4 +341,96 @@ object FileStreamSource {
def size: Int = map.size()
}
+
+ private[sql] trait FileStreamSourceCleaner {
+def clean(entry: FileEntry): Unit
+ }
+
+ private[sql] object FileStreamSourceCleaner {
+def apply(
+fileSystem: FileSystem,
+sourcePath: Path,
+option: FileStreamOptions,
+hadoopConf: Configuration): Option[FileStreamSourceCleaner] =
option.cleanSource match {
+ case CleanSourceMode.ARCHIVE =>
+require(option.sourceArchiveDir.isDefined)
+val path = new Path(option.sourceArchiveDir.get)
+val archiveFs = path.getFileSystem(hadoopConf)
+val qualifiedArchivePath = archiveFs.makeQualified(path)
+Some(new SourceFileArchiver(fileSystem, sourcePath, archiveFs,
qualifiedArchivePath))
+
+ case CleanSourceMode.DELETE =>
+Some(new SourceFileRemover(fileSystem))
+
+ case _ => None
+}
+ }
+
+ private[sql] class SourceFileArchiver(
+ fileSystem: FileSystem,
+ sourcePath: Path,
+ baseArchiveFileSystem: FileSystem,
+ baseArchivePath: Path) extends FileStreamSourceCleaner with Logging {
+assertParameters()
+
+private def assertParameters(): Unit = {
+ require(fileSystem.getUri == baseArchiveFileSystem.getUri, "Base archive
path is located " +
+s"on a different file system than the source files. source path:
$sourcePath" +
+s" / base archive path: $baseArchivePath")
+
+ /**
+ * FileStreamSource reads the files which one of below conditions is met:
+ * 1) file itself is matched with source path
+ * 2) parent directory is matched with source path
Review comment:
@zsxwing
Thanks for spending your time to revisit this! The condition is based on the
test suite in FileStreamSource, but for partitioned paths, yes that's missed.
Nice finding. I need to update the condition, or just remove the condition
documented there at all.
For `recursiveFileLookup`, it came later than the patch and I missed it. The
condition was formed in early this year, and recursiveFileLookup seemed to come
in mid this year.
Adding two cases, FileStreamSource can read any files under the source path,
which invalidates the depth check. There're three options to deal with this:
1) No pattern check and just try to rename. Log it if it fails to rename.
2) Disallow any path to be used as base archive path if the path matches the
source path (glob). After then we don't need to check the pattern.
3) Do pattern check before renaming, though it needs checking pattern per
file. We may optimize this a bit via grouping files per directory and check the
pattern with directory instead of individual files.
Which one (or couple of) would be the preferred approach?
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on a change in pull request #22952: [SPARK-20568][SS] Provide option to clean up completed files in streaming query
HeartSaVioR commented on a change in pull request #22952: [SPARK-20568][SS]
Provide option to clean up completed files in streaming query
URL: https://github.com/apache/spark/pull/22952#discussion_r347673657
##
File path:
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSource.scala
##
@@ -330,4 +341,96 @@ object FileStreamSource {
def size: Int = map.size()
}
+
+ private[sql] trait FileStreamSourceCleaner {
+def clean(entry: FileEntry): Unit
+ }
+
+ private[sql] object FileStreamSourceCleaner {
+def apply(
+fileSystem: FileSystem,
+sourcePath: Path,
+option: FileStreamOptions,
+hadoopConf: Configuration): Option[FileStreamSourceCleaner] =
option.cleanSource match {
+ case CleanSourceMode.ARCHIVE =>
+require(option.sourceArchiveDir.isDefined)
+val path = new Path(option.sourceArchiveDir.get)
+val archiveFs = path.getFileSystem(hadoopConf)
+val qualifiedArchivePath = archiveFs.makeQualified(path)
+Some(new SourceFileArchiver(fileSystem, sourcePath, archiveFs,
qualifiedArchivePath))
+
+ case CleanSourceMode.DELETE =>
+Some(new SourceFileRemover(fileSystem))
+
+ case _ => None
+}
+ }
+
+ private[sql] class SourceFileArchiver(
+ fileSystem: FileSystem,
+ sourcePath: Path,
+ baseArchiveFileSystem: FileSystem,
+ baseArchivePath: Path) extends FileStreamSourceCleaner with Logging {
+assertParameters()
+
+private def assertParameters(): Unit = {
+ require(fileSystem.getUri == baseArchiveFileSystem.getUri, "Base archive
path is located " +
+s"on a different file system than the source files. source path:
$sourcePath" +
+s" / base archive path: $baseArchivePath")
+
+ /**
+ * FileStreamSource reads the files which one of below conditions is met:
+ * 1) file itself is matched with source path
+ * 2) parent directory is matched with source path
Review comment:
@zsxwing
Thanks for spending your time to revisit this! The condition is based on the
test suite in FileStreamSource, but for partitioned paths, yes that's missed.
Nice finding. I need to update the condition, or just remove the condition
documented there at all.
For `recursiveFileLookup`, it came later than the patch and I missed it. The
condition was formed in early this year, and recursiveFileLookup seemed to come
in mid this year.
Adding two cases, FileStreamSource can read any files under the source path,
which invalidates the depth check. There're three options to deal with this:
1) No pattern check and just try to rename. Log it if it fails to rename.
(Caution! It doesn't prevent archived file to be added to source file again in
different directory.)
2) Disallow any path to be used as base archive path if the path matches the
source path (glob). After then we don't need to check the pattern.
3) Do pattern check before renaming, though it needs checking pattern per
file. We may optimize this a bit via grouping files per directory and check the
pattern with directory instead of individual files.
Which one (or couple of) would be the preferred approach?
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on a change in pull request #22952: [SPARK-20568][SS] Provide option to clean up completed files in streaming query
HeartSaVioR commented on a change in pull request #22952: [SPARK-20568][SS]
Provide option to clean up completed files in streaming query
URL: https://github.com/apache/spark/pull/22952#discussion_r347673657
##
File path:
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSource.scala
##
@@ -330,4 +341,96 @@ object FileStreamSource {
def size: Int = map.size()
}
+
+ private[sql] trait FileStreamSourceCleaner {
+def clean(entry: FileEntry): Unit
+ }
+
+ private[sql] object FileStreamSourceCleaner {
+def apply(
+fileSystem: FileSystem,
+sourcePath: Path,
+option: FileStreamOptions,
+hadoopConf: Configuration): Option[FileStreamSourceCleaner] =
option.cleanSource match {
+ case CleanSourceMode.ARCHIVE =>
+require(option.sourceArchiveDir.isDefined)
+val path = new Path(option.sourceArchiveDir.get)
+val archiveFs = path.getFileSystem(hadoopConf)
+val qualifiedArchivePath = archiveFs.makeQualified(path)
+Some(new SourceFileArchiver(fileSystem, sourcePath, archiveFs,
qualifiedArchivePath))
+
+ case CleanSourceMode.DELETE =>
+Some(new SourceFileRemover(fileSystem))
+
+ case _ => None
+}
+ }
+
+ private[sql] class SourceFileArchiver(
+ fileSystem: FileSystem,
+ sourcePath: Path,
+ baseArchiveFileSystem: FileSystem,
+ baseArchivePath: Path) extends FileStreamSourceCleaner with Logging {
+assertParameters()
+
+private def assertParameters(): Unit = {
+ require(fileSystem.getUri == baseArchiveFileSystem.getUri, "Base archive
path is located " +
+s"on a different file system than the source files. source path:
$sourcePath" +
+s" / base archive path: $baseArchivePath")
+
+ /**
+ * FileStreamSource reads the files which one of below conditions is met:
+ * 1) file itself is matched with source path
+ * 2) parent directory is matched with source path
Review comment:
@zsxwing
Thanks for spending your time to revisit this! The condition is based on the
test suite in FileStreamSource, but for partitioned paths, yes that's missed.
Nice finding. I need to update the condition, or just remove the condition
documented there at all.
For `recursiveFileLookup`, it came later than the patch and I missed it. The
condition was formed in early this year, and recursiveFileLookup seemed to come
in mid this year.
Adding two cases, FileStreamSource can read any files under the source path,
which invalidates the depth check. There're three options to deal with this:
1) No pattern check and just try to rename. Log it if it fails to rename.
(Caution! It doesn't prevent archived file to be added to source file again in
different directory.)
2) Disallow any path to be used as base archive path if the path matches the
source path (glob) - fail the query. After then we don't need to check the
pattern. If end users provide complicated glob path as source path, they also
may be puzzled how to not match.
3) Do pattern check before renaming, though it needs checking pattern per
file. We may optimize this a bit via grouping files per directory and check the
pattern with directory instead of individual files. It doesn't fail the query
so end users need to check whether the files are not cleaned up due to the
pattern check.
Which one (or couple of) would be the preferred approach?
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on a change in pull request #22952: [SPARK-20568][SS] Provide option to clean up completed files in streaming query
HeartSaVioR commented on a change in pull request #22952: [SPARK-20568][SS]
Provide option to clean up completed files in streaming query
URL: https://github.com/apache/spark/pull/22952#discussion_r347673657
##
File path:
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSource.scala
##
@@ -330,4 +341,96 @@ object FileStreamSource {
def size: Int = map.size()
}
+
+ private[sql] trait FileStreamSourceCleaner {
+def clean(entry: FileEntry): Unit
+ }
+
+ private[sql] object FileStreamSourceCleaner {
+def apply(
+fileSystem: FileSystem,
+sourcePath: Path,
+option: FileStreamOptions,
+hadoopConf: Configuration): Option[FileStreamSourceCleaner] =
option.cleanSource match {
+ case CleanSourceMode.ARCHIVE =>
+require(option.sourceArchiveDir.isDefined)
+val path = new Path(option.sourceArchiveDir.get)
+val archiveFs = path.getFileSystem(hadoopConf)
+val qualifiedArchivePath = archiveFs.makeQualified(path)
+Some(new SourceFileArchiver(fileSystem, sourcePath, archiveFs,
qualifiedArchivePath))
+
+ case CleanSourceMode.DELETE =>
+Some(new SourceFileRemover(fileSystem))
+
+ case _ => None
+}
+ }
+
+ private[sql] class SourceFileArchiver(
+ fileSystem: FileSystem,
+ sourcePath: Path,
+ baseArchiveFileSystem: FileSystem,
+ baseArchivePath: Path) extends FileStreamSourceCleaner with Logging {
+assertParameters()
+
+private def assertParameters(): Unit = {
+ require(fileSystem.getUri == baseArchiveFileSystem.getUri, "Base archive
path is located " +
+s"on a different file system than the source files. source path:
$sourcePath" +
+s" / base archive path: $baseArchivePath")
+
+ /**
+ * FileStreamSource reads the files which one of below conditions is met:
+ * 1) file itself is matched with source path
+ * 2) parent directory is matched with source path
Review comment:
@zsxwing
Thanks for spending your time to revisit this! The condition is based on the
test suite in FileStreamSource, but for partitioned paths, yes that's missed.
Nice finding. I need to update the condition, or just remove the condition
documented there at all.
For `recursiveFileLookup`, it came later than the patch and I missed it. The
condition was formed in early this year, and recursiveFileLookup seemed to come
in mid this year.
Adding two cases, FileStreamSource can read any files under the source path,
which invalidates the depth check. There're three options to deal with this:
1) No pattern check and just try to rename. Log it if it fails to rename.
(Caution! It doesn't prevent archived file to be added to source file again in
different directory.)
2) Disallow any path to be used as base archive path if the path matches the
source path (glob) - here "disallow" means fail the query. After then we don't
need to check the pattern. If end users provide complicated glob path as source
path, they also may be puzzled how to not match.
3) Do pattern check before renaming, though it needs checking pattern per
file. We may optimize this a bit via grouping files per directory and check the
pattern with directory instead of individual files. It doesn't fail the query
so end users need to check whether the files are not cleaned up due to the
pattern check.
Which one (or couple of) would be the preferred approach?
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on a change in pull request #22952: [SPARK-20568][SS] Provide option to clean up completed files in streaming query
HeartSaVioR commented on a change in pull request #22952: [SPARK-20568][SS]
Provide option to clean up completed files in streaming query
URL: https://github.com/apache/spark/pull/22952#discussion_r347673657
##
File path:
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSource.scala
##
@@ -330,4 +341,96 @@ object FileStreamSource {
def size: Int = map.size()
}
+
+ private[sql] trait FileStreamSourceCleaner {
+def clean(entry: FileEntry): Unit
+ }
+
+ private[sql] object FileStreamSourceCleaner {
+def apply(
+fileSystem: FileSystem,
+sourcePath: Path,
+option: FileStreamOptions,
+hadoopConf: Configuration): Option[FileStreamSourceCleaner] =
option.cleanSource match {
+ case CleanSourceMode.ARCHIVE =>
+require(option.sourceArchiveDir.isDefined)
+val path = new Path(option.sourceArchiveDir.get)
+val archiveFs = path.getFileSystem(hadoopConf)
+val qualifiedArchivePath = archiveFs.makeQualified(path)
+Some(new SourceFileArchiver(fileSystem, sourcePath, archiveFs,
qualifiedArchivePath))
+
+ case CleanSourceMode.DELETE =>
+Some(new SourceFileRemover(fileSystem))
+
+ case _ => None
+}
+ }
+
+ private[sql] class SourceFileArchiver(
+ fileSystem: FileSystem,
+ sourcePath: Path,
+ baseArchiveFileSystem: FileSystem,
+ baseArchivePath: Path) extends FileStreamSourceCleaner with Logging {
+assertParameters()
+
+private def assertParameters(): Unit = {
+ require(fileSystem.getUri == baseArchiveFileSystem.getUri, "Base archive
path is located " +
+s"on a different file system than the source files. source path:
$sourcePath" +
+s" / base archive path: $baseArchivePath")
+
+ /**
+ * FileStreamSource reads the files which one of below conditions is met:
+ * 1) file itself is matched with source path
+ * 2) parent directory is matched with source path
Review comment:
@zsxwing
Thanks for spending your time to revisit this! The condition is based on the
test suite in FileStreamSource, but for partitioned paths, yes that's missed.
Nice finding. I need to update the condition, or just remove the condition
documented there at all.
For `recursiveFileLookup`, it came later than the patch and I missed it. The
condition was formed in early this year, and recursiveFileLookup seemed to come
in mid this year.
Adding two cases, FileStreamSource can read any files under the source path,
which invalidates the depth check. There're three options to deal with this:
1) No pattern check and just try to rename. Log it if it fails to rename.
(Caution! It doesn't prevent archived file to be added to source file again in
different directory.)
2) Disallow any path to be used as base archive path if the path matches the
source path (glob) - here "disallow" means fail the query. After then we don't
need to check the pattern. If end users provide complicated glob path as source
path, they also may be puzzled how to not match, but not sure they would really
want to set the path be complicated in production.
3) Do pattern check before renaming, though it needs checking pattern per
file. We may optimize this a bit via grouping files per directory and check the
pattern with directory instead of individual files. It doesn't fail the query
so end users need to check whether the files are not cleaned up due to the
pattern check.
Which one (or couple of) would be the preferred approach?
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
