You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by zsxwing <gi...@git.apache.org> on 2016/03/10 02:42:44 UTC
[GitHub] spark pull request: [SPARK-13791][SQL]Add MetadataLog and HDFSMeta...
GitHub user zsxwing opened a pull request:
https://github.com/apache/spark/pull/11625
[SPARK-13791][SQL]Add MetadataLog and HDFSMetadataLog
## What changes were proposed in this pull request?
- Add a MetadataLog interface for metadata reliably storage.
- Add HDFSMetadataLog as a MetadataLog implementation based on HDFS.
- Update FileStreamSource to use HDFSMetadataLog instead of managing metadata by itself.
## How was this patch tested?
unit tests
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/zsxwing/spark metadata-log
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/spark/pull/11625.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #11625
----
commit c32cd963ba9bdd1ec0eec6b74bdbf361c00a2bf9
Author: Shixiong Zhu <sh...@databricks.com>
Date: 2016-03-08T22:40:14Z
Add MetadataLog and HDFSMetadataLog
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13791][SQL]Add MetadataLog and HDFSMeta...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11625#issuecomment-196530048
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/53112/
Test FAILed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13791][SQL]Add MetadataLog and HDFSMeta...
Posted by marmbrus <gi...@git.apache.org>.
Github user marmbrus commented on a diff in the pull request:
https://github.com/apache/spark/pull/11625#discussion_r55741426
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/MetadataLog.scala ---
@@ -0,0 +1,55 @@
+/*
+* Licensed to the Apache Software Foundation (ASF) under one or more
+* contributor license agreements. See the NOTICE file distributed with
+* this work for additional information regarding copyright ownership.
+* The ASF licenses this file to You under the Apache License, Version 2.0
+* (the "License"); you may not use this file except in compliance with
+* the License. You may obtain a copy of the License at
+*
+* http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing, software
+* distributed under the License is distributed on an "AS IS" BASIS,
+* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+* See the License for the specific language governing permissions and
+* limitations under the License.
+*/
+
+package org.apache.spark.sql.execution.streaming
+
+/**
+ * A general MetadataLog that supports the following features:
+ *
+ * - Allow the user to store a metadata object for each batch.
+ * - Allow the user to query the latest batch id.
+ * - Allow the user to query the metadata object of a specified batch id.
+ * - Allow the user to query metadata objects in a range of batch ids.
+ */
+trait MetadataLog[T] {
+
+ /**
+ * Store the metadata for the specified batchId. If batchId's metatdata has already been stored,
+ * this method does nothing.
+ */
+ def add(batchId: Long, metadata: T): Unit
+
+ /**
+ * Return the metadata for the specified batchId if it's stored. Otherwise, return None.
+ */
+ def get(batchId: Long): Option[T]
+
+ /**
+ * Return metadata for batches between startId (inclusive) and endId (inclusive).
--- End diff --
What does `startId = None` mean?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13791][SQL]Add MetadataLog and HDFSMeta...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11625#issuecomment-196513774
**[Test build #53103 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/53103/consoleFull)** for PR 11625 at commit [`4c27c6e`](https://github.com/apache/spark/commit/4c27c6eb2ba17cebacee3a77b2ce5643e6da0329).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13791][SQL]Add MetadataLog and HDFSMeta...
Posted by marmbrus <gi...@git.apache.org>.
Github user marmbrus commented on a diff in the pull request:
https://github.com/apache/spark/pull/11625#discussion_r55740852
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/HDFSMetadataLog.scala ---
@@ -0,0 +1,168 @@
+/*
+* Licensed to the Apache Software Foundation (ASF) under one or more
+* contributor license agreements. See the NOTICE file distributed with
+* this work for additional information regarding copyright ownership.
+* The ASF licenses this file to You under the Apache License, Version 2.0
+* (the "License"); you may not use this file except in compliance with
+* the License. You may obtain a copy of the License at
+*
+* http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing, software
+* distributed under the License is distributed on an "AS IS" BASIS,
+* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+* See the License for the specific language governing permissions and
+* limitations under the License.
+*/
+
+package org.apache.spark.sql.execution.streaming
+
+import java.io.{DataOutput, EOFException, IOException}
+import java.nio.ByteBuffer
+
+import scala.reflect.ClassTag
+
+import org.apache.hadoop.fs.{Path, PathFilter}
+
+import org.apache.spark.SparkException
+import org.apache.spark.serializer.JavaSerializer
+import org.apache.spark.sql.SQLContext
+import org.apache.spark.util.Utils
+
+/**
+ * A [[MetadataLog]] implementation based on HDFS. [[HDFSMetadataLog]] uses the specified `path`
+ * as the metadata storage.
+ *
+ * There should only be one [[HDFSMetadataLog]] using `path` at the same time. [[HDFSMetadataLog]]
+ * uses a file ".lock" in the directory to make sure there is always only one user using `path`.
+ * When [[HDFSMetadataLog]] is created, it will create a ".lock" file. If this step fails,
+ * [[HDFSMetadataLog]] will throw an exception saying there is someone using the same directory.
+ * When [[HDFSMetadataLog]] is stopped, it will delete the ".lock" file.
+ *
+ * However, in extreme case, e.g., power outage, [[HDFSMetadataLog]] won't be able to delete ".lock"
+ * file. Then [[HDFSMetadataLog]] cannot be created even if no one is using the `path`. In such
+ * case, the user has to delete the lock file in the exception message to restart their application.
+ *
+ * Note: [[HDFSMetadataLog]] doesn't support S3-like file systems as they don't guarantee listing
+ * files in a directory always shows the latest files.
+ */
+class HDFSMetadataLog[T: ClassTag](sqlContext: SQLContext, path: String) extends MetadataLog[T] {
+
+ private val metadataPath = new Path(path)
+ private val fs = metadataPath.getFileSystem(sqlContext.sparkContext.hadoopConfiguration)
+ if (!fs.exists(metadataPath)) {
+ fs.mkdirs(metadataPath)
+ }
+
+ tryAcquireLock()
+
+ /**
+ * A `PathFilter` to filter only batch files
+ */
+ private val batchFilesFilter = new PathFilter {
+ override def accept(path: Path): Boolean = try {
+ path.getName.toLong
+ true
+ } catch {
+ case _: NumberFormatException => false
+ }
+ }
+
+ private val serializer = new JavaSerializer(sqlContext.sparkContext.conf).newInstance()
+
+ private def tryAcquireLock(): Unit = {
+ val lockFile = new Path(metadataPath, ".lock")
+ fs.deleteOnExit(lockFile)
+ try {
+ // Must set overwrite to false otherwise HDFS allows multiple clients to create the same file.
+ // When overwrite is false, the latter one will receive FileAlreadyExistsException
+ fs.create(lockFile, false).close()
+ } catch {
+ case e: IOException =>
+ throw new SparkException(s"There are another user using $path. Please check that. " +
+ s"If nobody is using it, please delete ${lockFile}", e)
+ }
+ }
+
+ private def batchFile(batchId: Long): Path = {
+ new Path(metadataPath, batchId.toString)
+ }
+
+ override def add(batchId: Long, metadata: T): Unit = {
+ get(batchId).getOrElse {
+ // Only write metadata when the batch has not yet been written.
+ val buffer = serializer.serialize(metadata)
+ val output = try {
+ fs.create(batchFile(batchId))
+ } catch {
+ case e: IOException if "java.lang.InterruptedException" == e.getMessage =>
--- End diff --
Should you be checking the `cause` instead of `getMessage`?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13791][SQL]Add MetadataLog and HDFSMeta...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11625#issuecomment-196525235
**[Test build #53112 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/53112/consoleFull)** for PR 11625 at commit [`4c27c6e`](https://github.com/apache/spark/commit/4c27c6eb2ba17cebacee3a77b2ce5643e6da0329).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13791][SQL]Add MetadataLog and HDFSMeta...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11625#issuecomment-196600745
Merged build finished. Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13791][SQL]Add MetadataLog and HDFSMeta...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11625#issuecomment-196600748
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/53137/
Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13791][SQL]Add MetadataLog and HDFSMeta...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11625#issuecomment-196575295
**[Test build #53137 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/53137/consoleFull)** for PR 11625 at commit [`1c82e56`](https://github.com/apache/spark/commit/1c82e5681c35e1a169a9e3f49720adaa19462cce).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13791][SQL]Add MetadataLog and HDFSMeta...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11625#issuecomment-194641031
**[Test build #52801 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/52801/consoleFull)** for PR 11625 at commit [`c32cd96`](https://github.com/apache/spark/commit/c32cd963ba9bdd1ec0eec6b74bdbf361c00a2bf9).
* This patch **fails Spark unit tests**.
* This patch merges cleanly.
* This patch adds the following public classes _(experimental)_:
* `class HDFSMetadataLog[T: ClassTag](sqlContext: SQLContext, path: String) extends MetadataLog[T] `
* `trait MetadataLog[T] `
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13791][SQL]Add MetadataLog and HDFSMeta...
Posted by marmbrus <gi...@git.apache.org>.
Github user marmbrus commented on a diff in the pull request:
https://github.com/apache/spark/pull/11625#discussion_r55741318
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/HDFSMetadataLog.scala ---
@@ -0,0 +1,168 @@
+/*
+* Licensed to the Apache Software Foundation (ASF) under one or more
+* contributor license agreements. See the NOTICE file distributed with
+* this work for additional information regarding copyright ownership.
+* The ASF licenses this file to You under the Apache License, Version 2.0
+* (the "License"); you may not use this file except in compliance with
+* the License. You may obtain a copy of the License at
+*
+* http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing, software
+* distributed under the License is distributed on an "AS IS" BASIS,
+* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+* See the License for the specific language governing permissions and
+* limitations under the License.
+*/
+
+package org.apache.spark.sql.execution.streaming
+
+import java.io.{DataOutput, EOFException, IOException}
+import java.nio.ByteBuffer
+
+import scala.reflect.ClassTag
+
+import org.apache.hadoop.fs.{Path, PathFilter}
+
+import org.apache.spark.SparkException
+import org.apache.spark.serializer.JavaSerializer
+import org.apache.spark.sql.SQLContext
+import org.apache.spark.util.Utils
+
+/**
+ * A [[MetadataLog]] implementation based on HDFS. [[HDFSMetadataLog]] uses the specified `path`
+ * as the metadata storage.
+ *
+ * There should only be one [[HDFSMetadataLog]] using `path` at the same time. [[HDFSMetadataLog]]
+ * uses a file ".lock" in the directory to make sure there is always only one user using `path`.
+ * When [[HDFSMetadataLog]] is created, it will create a ".lock" file. If this step fails,
+ * [[HDFSMetadataLog]] will throw an exception saying there is someone using the same directory.
+ * When [[HDFSMetadataLog]] is stopped, it will delete the ".lock" file.
+ *
+ * However, in extreme case, e.g., power outage, [[HDFSMetadataLog]] won't be able to delete ".lock"
+ * file. Then [[HDFSMetadataLog]] cannot be created even if no one is using the `path`. In such
+ * case, the user has to delete the lock file in the exception message to restart their application.
+ *
+ * Note: [[HDFSMetadataLog]] doesn't support S3-like file systems as they don't guarantee listing
+ * files in a directory always shows the latest files.
+ */
+class HDFSMetadataLog[T: ClassTag](sqlContext: SQLContext, path: String) extends MetadataLog[T] {
+
+ private val metadataPath = new Path(path)
+ private val fs = metadataPath.getFileSystem(sqlContext.sparkContext.hadoopConfiguration)
+ if (!fs.exists(metadataPath)) {
+ fs.mkdirs(metadataPath)
+ }
+
+ tryAcquireLock()
+
+ /**
+ * A `PathFilter` to filter only batch files
+ */
+ private val batchFilesFilter = new PathFilter {
+ override def accept(path: Path): Boolean = try {
+ path.getName.toLong
+ true
+ } catch {
+ case _: NumberFormatException => false
+ }
+ }
+
+ private val serializer = new JavaSerializer(sqlContext.sparkContext.conf).newInstance()
+
+ private def tryAcquireLock(): Unit = {
+ val lockFile = new Path(metadataPath, ".lock")
+ fs.deleteOnExit(lockFile)
+ try {
+ // Must set overwrite to false otherwise HDFS allows multiple clients to create the same file.
+ // When overwrite is false, the latter one will receive FileAlreadyExistsException
+ fs.create(lockFile, false).close()
+ } catch {
+ case e: IOException =>
+ throw new SparkException(s"There are another user using $path. Please check that. " +
+ s"If nobody is using it, please delete ${lockFile}", e)
+ }
+ }
+
+ private def batchFile(batchId: Long): Path = {
+ new Path(metadataPath, batchId.toString)
+ }
+
+ override def add(batchId: Long, metadata: T): Unit = {
+ get(batchId).getOrElse {
+ // Only write metadata when the batch has not yet been written.
+ val buffer = serializer.serialize(metadata)
+ val output = try {
+ fs.create(batchFile(batchId))
+ } catch {
+ case e: IOException if "java.lang.InterruptedException" == e.getMessage =>
+ // create may convert InterruptedException to IOException. Let's convert it back to
+ // InterruptedException so that this failure won't crash StreamExecution
+ throw new InterruptedException("Creating file is interrupted")
+ }
+ try {
+ output.writeInt(buffer.remaining())
+ Utils.writeByteBuffer(buffer, output: DataOutput)
+ } finally {
+ output.close()
+ }
+ }
+ }
+
+ override def get(batchId: Long): Option[T] = {
+ val batchMetadataFile = batchFile(batchId)
+ if (fs.exists(batchMetadataFile)) {
+ val input = fs.open(batchMetadataFile)
+ try {
+ val size = input.readInt()
+ val bytes = new Array[Byte](size)
+ val readSize = input.read(bytes)
+ if (readSize == size) {
+ Some(serializer.deserialize[T](ByteBuffer.wrap(bytes)))
+ } else {
+ None
+ }
+ } catch {
+ case _: EOFException =>
+ // The file is corrupted, so return None
--- End diff --
It would be good to add some comments about how we define file corruption.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13791][SQL]Add MetadataLog and HDFSMeta...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11625#issuecomment-196482805
**[Test build #53083 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/53083/consoleFull)** for PR 11625 at commit [`7a52adc`](https://github.com/apache/spark/commit/7a52adc35d7e7100bc9271d54196b7eb85f5015b).
* This patch **fails Spark unit tests**.
* This patch merges cleanly.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13791][SQL]Add MetadataLog and HDFSMeta...
Posted by zsxwing <gi...@git.apache.org>.
Github user zsxwing commented on a diff in the pull request:
https://github.com/apache/spark/pull/11625#discussion_r55739237
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/HDFSMetadataLog.scala ---
@@ -0,0 +1,168 @@
+/*
+* Licensed to the Apache Software Foundation (ASF) under one or more
+* contributor license agreements. See the NOTICE file distributed with
+* this work for additional information regarding copyright ownership.
+* The ASF licenses this file to You under the Apache License, Version 2.0
+* (the "License"); you may not use this file except in compliance with
+* the License. You may obtain a copy of the License at
+*
+* http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing, software
+* distributed under the License is distributed on an "AS IS" BASIS,
+* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+* See the License for the specific language governing permissions and
+* limitations under the License.
+*/
+
+package org.apache.spark.sql.execution.streaming
+
+import java.io.{DataOutput, EOFException, IOException}
+import java.nio.ByteBuffer
+
+import scala.reflect.ClassTag
+
+import org.apache.hadoop.fs.{Path, PathFilter}
+
+import org.apache.spark.SparkException
+import org.apache.spark.serializer.JavaSerializer
+import org.apache.spark.sql.SQLContext
+import org.apache.spark.util.Utils
+
+/**
+ * A [[MetadataLog]] implementation based on HDFS. [[HDFSMetadataLog]] uses the specified `path`
+ * as the metadata storage.
+ *
+ * There should only be one [[HDFSMetadataLog]] using `path` at the same time. [[HDFSMetadataLog]]
+ * uses a file ".lock" in the directory to make sure there is always only one user using `path`.
+ * When [[HDFSMetadataLog]] is created, it will create a ".lock" file. If this step fails,
+ * [[HDFSMetadataLog]] will throw an exception saying there is someone using the same directory.
+ * When [[HDFSMetadataLog]] is stopped, it will delete the ".lock" file.
+ *
+ * However, in extreme case, e.g., power outage, [[HDFSMetadataLog]] won't be able to delete ".lock"
+ * file. Then [[HDFSMetadataLog]] cannot be created even if no one is using the `path`. In such
+ * case, the user has to delete the lock file in the exception message to restart their application.
+ *
+ * Note: [[HDFSMetadataLog]] doesn't support S3-like file systems as they don't guarantee listing
+ * files in a directory always shows the latest files.
+ */
+class HDFSMetadataLog[T: ClassTag](sqlContext: SQLContext, path: String) extends MetadataLog[T] {
+
+ private val metadataPath = new Path(path)
+ private val fs = metadataPath.getFileSystem(sqlContext.sparkContext.hadoopConfiguration)
+ if (!fs.exists(metadataPath)) {
+ fs.mkdirs(metadataPath)
+ }
+
+ tryAcquireLock()
+
+ /**
+ * A `PathFilter` to filter only batch files
+ */
+ private val batchFilesFilter = new PathFilter {
+ override def accept(path: Path): Boolean = try {
+ path.getName.toLong
+ true
+ } catch {
+ case _: NumberFormatException => false
+ }
+ }
+
+ private val serializer = new JavaSerializer(sqlContext.sparkContext.conf).newInstance()
+
+ private def tryAcquireLock(): Unit = {
--- End diff --
As the semantics of add is `If batchId's metatdata has already been stored, this method does nothing.`, if the writer A sees a file created by the other writer B, it won't write the file. So A won't fail and will use the metadata written by B.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13791][SQL]Add MetadataLog and HDFSMeta...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/spark/pull/11625
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13791][SQL]Add MetadataLog and HDFSMeta...
Posted by marmbrus <gi...@git.apache.org>.
Github user marmbrus commented on a diff in the pull request:
https://github.com/apache/spark/pull/11625#discussion_r55741277
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/HDFSMetadataLog.scala ---
@@ -0,0 +1,168 @@
+/*
+* Licensed to the Apache Software Foundation (ASF) under one or more
+* contributor license agreements. See the NOTICE file distributed with
+* this work for additional information regarding copyright ownership.
+* The ASF licenses this file to You under the Apache License, Version 2.0
+* (the "License"); you may not use this file except in compliance with
+* the License. You may obtain a copy of the License at
+*
+* http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing, software
+* distributed under the License is distributed on an "AS IS" BASIS,
+* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+* See the License for the specific language governing permissions and
+* limitations under the License.
+*/
+
+package org.apache.spark.sql.execution.streaming
+
+import java.io.{DataOutput, EOFException, IOException}
+import java.nio.ByteBuffer
+
+import scala.reflect.ClassTag
+
+import org.apache.hadoop.fs.{Path, PathFilter}
+
+import org.apache.spark.SparkException
+import org.apache.spark.serializer.JavaSerializer
+import org.apache.spark.sql.SQLContext
+import org.apache.spark.util.Utils
+
+/**
+ * A [[MetadataLog]] implementation based on HDFS. [[HDFSMetadataLog]] uses the specified `path`
+ * as the metadata storage.
+ *
+ * There should only be one [[HDFSMetadataLog]] using `path` at the same time. [[HDFSMetadataLog]]
+ * uses a file ".lock" in the directory to make sure there is always only one user using `path`.
+ * When [[HDFSMetadataLog]] is created, it will create a ".lock" file. If this step fails,
+ * [[HDFSMetadataLog]] will throw an exception saying there is someone using the same directory.
+ * When [[HDFSMetadataLog]] is stopped, it will delete the ".lock" file.
+ *
+ * However, in extreme case, e.g., power outage, [[HDFSMetadataLog]] won't be able to delete ".lock"
+ * file. Then [[HDFSMetadataLog]] cannot be created even if no one is using the `path`. In such
+ * case, the user has to delete the lock file in the exception message to restart their application.
+ *
+ * Note: [[HDFSMetadataLog]] doesn't support S3-like file systems as they don't guarantee listing
+ * files in a directory always shows the latest files.
+ */
+class HDFSMetadataLog[T: ClassTag](sqlContext: SQLContext, path: String) extends MetadataLog[T] {
+
+ private val metadataPath = new Path(path)
+ private val fs = metadataPath.getFileSystem(sqlContext.sparkContext.hadoopConfiguration)
+ if (!fs.exists(metadataPath)) {
+ fs.mkdirs(metadataPath)
+ }
+
+ tryAcquireLock()
+
+ /**
+ * A `PathFilter` to filter only batch files
+ */
+ private val batchFilesFilter = new PathFilter {
+ override def accept(path: Path): Boolean = try {
+ path.getName.toLong
+ true
+ } catch {
+ case _: NumberFormatException => false
+ }
+ }
+
+ private val serializer = new JavaSerializer(sqlContext.sparkContext.conf).newInstance()
+
+ private def tryAcquireLock(): Unit = {
+ val lockFile = new Path(metadataPath, ".lock")
+ fs.deleteOnExit(lockFile)
+ try {
+ // Must set overwrite to false otherwise HDFS allows multiple clients to create the same file.
+ // When overwrite is false, the latter one will receive FileAlreadyExistsException
+ fs.create(lockFile, false).close()
+ } catch {
+ case e: IOException =>
+ throw new SparkException(s"There are another user using $path. Please check that. " +
+ s"If nobody is using it, please delete ${lockFile}", e)
+ }
+ }
+
+ private def batchFile(batchId: Long): Path = {
+ new Path(metadataPath, batchId.toString)
+ }
+
+ override def add(batchId: Long, metadata: T): Unit = {
+ get(batchId).getOrElse {
+ // Only write metadata when the batch has not yet been written.
+ val buffer = serializer.serialize(metadata)
+ val output = try {
+ fs.create(batchFile(batchId))
+ } catch {
+ case e: IOException if "java.lang.InterruptedException" == e.getMessage =>
+ // create may convert InterruptedException to IOException. Let's convert it back to
+ // InterruptedException so that this failure won't crash StreamExecution
+ throw new InterruptedException("Creating file is interrupted")
+ }
+ try {
+ output.writeInt(buffer.remaining())
+ Utils.writeByteBuffer(buffer, output: DataOutput)
+ } finally {
+ output.close()
+ }
+ }
+ }
+
+ override def get(batchId: Long): Option[T] = {
+ val batchMetadataFile = batchFile(batchId)
+ if (fs.exists(batchMetadataFile)) {
+ val input = fs.open(batchMetadataFile)
+ try {
+ val size = input.readInt()
+ val bytes = new Array[Byte](size)
+ val readSize = input.read(bytes)
+ if (readSize == size) {
+ Some(serializer.deserialize[T](ByteBuffer.wrap(bytes)))
+ } else {
+ None
+ }
+ } catch {
+ case _: EOFException =>
+ // The file is corrupted, so return None
--- End diff --
I'm not sure if this is a big deal, but this does not handle the case where there is garbage at the end of the file, right?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13791][SQL]Add MetadataLog and HDFSMeta...
Posted by marmbrus <gi...@git.apache.org>.
Github user marmbrus commented on a diff in the pull request:
https://github.com/apache/spark/pull/11625#discussion_r55738627
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/HDFSMetadataLog.scala ---
@@ -0,0 +1,168 @@
+/*
+* Licensed to the Apache Software Foundation (ASF) under one or more
+* contributor license agreements. See the NOTICE file distributed with
+* this work for additional information regarding copyright ownership.
+* The ASF licenses this file to You under the Apache License, Version 2.0
+* (the "License"); you may not use this file except in compliance with
+* the License. You may obtain a copy of the License at
+*
+* http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing, software
+* distributed under the License is distributed on an "AS IS" BASIS,
+* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+* See the License for the specific language governing permissions and
+* limitations under the License.
+*/
+
+package org.apache.spark.sql.execution.streaming
+
+import java.io.{DataOutput, EOFException, IOException}
+import java.nio.ByteBuffer
+
+import scala.reflect.ClassTag
+
+import org.apache.hadoop.fs.{Path, PathFilter}
+
+import org.apache.spark.SparkException
+import org.apache.spark.serializer.JavaSerializer
+import org.apache.spark.sql.SQLContext
+import org.apache.spark.util.Utils
+
+/**
+ * A [[MetadataLog]] implementation based on HDFS. [[HDFSMetadataLog]] uses the specified `path`
+ * as the metadata storage.
+ *
+ * There should only be one [[HDFSMetadataLog]] using `path` at the same time. [[HDFSMetadataLog]]
+ * uses a file ".lock" in the directory to make sure there is always only one user using `path`.
+ * When [[HDFSMetadataLog]] is created, it will create a ".lock" file. If this step fails,
+ * [[HDFSMetadataLog]] will throw an exception saying there is someone using the same directory.
+ * When [[HDFSMetadataLog]] is stopped, it will delete the ".lock" file.
+ *
+ * However, in extreme case, e.g., power outage, [[HDFSMetadataLog]] won't be able to delete ".lock"
+ * file. Then [[HDFSMetadataLog]] cannot be created even if no one is using the `path`. In such
+ * case, the user has to delete the lock file in the exception message to restart their application.
+ *
+ * Note: [[HDFSMetadataLog]] doesn't support S3-like file systems as they don't guarantee listing
+ * files in a directory always shows the latest files.
+ */
+class HDFSMetadataLog[T: ClassTag](sqlContext: SQLContext, path: String) extends MetadataLog[T] {
+
+ private val metadataPath = new Path(path)
+ private val fs = metadataPath.getFileSystem(sqlContext.sparkContext.hadoopConfiguration)
+ if (!fs.exists(metadataPath)) {
+ fs.mkdirs(metadataPath)
+ }
+
+ tryAcquireLock()
+
+ /**
+ * A `PathFilter` to filter only batch files
+ */
+ private val batchFilesFilter = new PathFilter {
+ override def accept(path: Path): Boolean = try {
+ path.getName.toLong
+ true
+ } catch {
+ case _: NumberFormatException => false
+ }
+ }
+
+ private val serializer = new JavaSerializer(sqlContext.sparkContext.conf).newInstance()
+
+ private def tryAcquireLock(): Unit = {
--- End diff --
What happens if you die while you are holding the lock? It seems like your streaming job will be unable to restart without human intervention. Is there a reason that we can't detect problems when writing a new log entry?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13791][SQL]Add MetadataLog and HDFSMeta...
Posted by marmbrus <gi...@git.apache.org>.
Github user marmbrus commented on a diff in the pull request:
https://github.com/apache/spark/pull/11625#discussion_r56094033
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/HDFSMetadataLog.scala ---
@@ -0,0 +1,196 @@
+/*
+* Licensed to the Apache Software Foundation (ASF) under one or more
+* contributor license agreements. See the NOTICE file distributed with
+* this work for additional information regarding copyright ownership.
+* The ASF licenses this file to You under the Apache License, Version 2.0
+* (the "License"); you may not use this file except in compliance with
+* the License. You may obtain a copy of the License at
+*
+* http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing, software
+* distributed under the License is distributed on an "AS IS" BASIS,
+* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+* See the License for the specific language governing permissions and
+* limitations under the License.
+*/
+
+package org.apache.spark.sql.execution.streaming
+
+import java.io.{FileNotFoundException, IOException}
+import java.nio.ByteBuffer
+import java.util.{ConcurrentModificationException, EnumSet}
+
+import scala.reflect.ClassTag
+
+import org.apache.commons.io.IOUtils
+import org.apache.hadoop.fs._
+import org.apache.hadoop.fs.permission.FsPermission
+
+import org.apache.spark.network.util.JavaUtils
+import org.apache.spark.serializer.JavaSerializer
+import org.apache.spark.sql.SQLContext
+
+/**
+ * A [[MetadataLog]] implementation based on HDFS. [[HDFSMetadataLog]] uses the specified `path`
+ * as the metadata storage.
+ *
+ * When writing a new batch, [[HDFSMetadataLog]] will firstly write to a temp file and then rename
+ * it to the final batch file. If the rename step fails, there must be multiple writers and only
+ * one of them will succeed and the others will fail.
+ *
+ * Note: [[HDFSMetadataLog]] doesn't support S3-like file systems as they don't guarantee listing
+ * files in a directory always shows the latest files.
+ */
+class HDFSMetadataLog[T: ClassTag](sqlContext: SQLContext, path: String) extends MetadataLog[T] {
+
+ private val metadataPath = new Path(path)
+
+ private val fc =
+ if (metadataPath.toUri.getScheme == null) {
+ FileContext.getFileContext(sqlContext.sparkContext.hadoopConfiguration)
+ } else {
+ FileContext.getFileContext(metadataPath.toUri, sqlContext.sparkContext.hadoopConfiguration)
+ }
+
+ if (!fc.util().exists(metadataPath)) {
+ fc.mkdir(metadataPath, FsPermission.getDirDefault, true)
+ }
+
+ /**
+ * A `PathFilter` to filter only batch files
+ */
+ private val batchFilesFilter = new PathFilter {
+ override def accept(path: Path): Boolean = try {
+ path.getName.toLong
+ true
+ } catch {
+ case _: NumberFormatException => false
+ }
+ }
+
+ private val serializer = new JavaSerializer(sqlContext.sparkContext.conf).newInstance()
+
+ private def batchFile(batchId: Long): Path = {
+ new Path(metadataPath, batchId.toString)
+ }
+
+ override def add(batchId: Long, metadata: T): Unit = {
+ get(batchId).map(_ => false).getOrElse {
+ // Only write metadata when the batch has not yet been written.
+ val buffer = serializer.serialize(metadata)
+ try {
+ writeBatch(batchId, JavaUtils.bufferToArray(buffer))
+ true
+ } catch {
+ case e: IOException if "java.lang.InterruptedException" == e.getMessage =>
+ // create may convert InterruptedException to IOException. Let's convert it back to
+ // InterruptedException so that this failure won't crash StreamExecution
+ throw new InterruptedException("Creating file is interrupted")
+ }
+ }
+ }
+
+ /**
+ * Write a batch to a temp file then rename it to the batch file.
+ *
+ * There may be multiple [[HDFSMetadataLog]] using the same metadata path. Although it is not a
+ * valid behavior, we still need to prevent it from destroying the files.
+ */
+ private def writeBatch(batchId: Long, bytes: Array[Byte]): Unit = {
+ // Use nextId to create a temp file
+ var nextId = 0
+ while (true) {
+ val tempPath = new Path(metadataPath, s".${batchId}_$nextId.tmp")
+ fc.deleteOnExit(tempPath)
+ try {
+ val output = fc.create(tempPath, EnumSet.of(CreateFlag.CREATE))
+ try {
+ output.write(bytes)
+ } finally {
+ output.close()
+ }
+ try {
+ // Try to commit the batch
+ // It will fail if there is an existing file (someone has committed the batch)
+ fc.rename(tempPath, batchFile(batchId), Options.Rename.NONE)
+ return
+ } catch {
+ case e: IOException if isFileAlreadyExistsException(e) =>
+ // If "rename" fails, it means some other "HDFSMetadataLog" has committed the batch.
+ // So throw an exception to tell the user this is not a valid behavior.
+ throw new ConcurrentModificationException(
+ s"Multiple HDFSMetadataLog are using $path", e)
+ case e: FileNotFoundException =>
+ // Sometimes, "create" will succeed when multiple writers are calling it at the same
+ // time. However, only one writer can call "rename" successfully, others will get
+ // FileNotFoundException because the first writer has removed it.
+ throw new ConcurrentModificationException(
+ s"Multiple HDFSMetadataLog are using $path", e)
+ }
+ } catch {
+ case e: IOException if isFileAlreadyExistsException(e) =>
+ // Failed to create "tempPath". There are two cases:
+ // 1. Someone is creating "tempPath" too.
+ // 2. This is a restart. "tempPath" has already been created but not moved to the final
+ // batch file (not committed).
+ //
+ // For both cases, the batch has not yet been committed. So we can retry it.
+ //
+ // Note: there is a potential risk here: if HDFSMetadataLog A is running, people can use
+ // the same metadata path to create "HDFSMetadataLog" and fail A. However, this is not a
+ // big problem because it requires the attacker must have the permission to write the
+ // metadata path. In addition, the old Streaming also have this issue, people can create
+ // malicious checkpoint files to crash a Streaming application too.
+ nextId += 1
+ }
+ }
+ }
+
+ private def isFileAlreadyExistsException(e: IOException): Boolean = {
+ e.isInstanceOf[FileAlreadyExistsException] ||
+ // Old Hadoop versions don't throw FileAlreadyExistsException. Although it's fixed in
+ // HADOOP-9361, we still need to support old Hadoop versions.
+ (e.getMessage != null && e.getMessage.startsWith("File already exists: "))
+ }
+
+ override def get(batchId: Long): Option[T] = {
+ val batchMetadataFile = batchFile(batchId)
+ if (fc.util().exists(batchMetadataFile)) {
+ val input = fc.open(batchMetadataFile)
+ val bytes = IOUtils.toByteArray(input)
+ Some(serializer.deserialize[T](ByteBuffer.wrap(bytes)))
+ } else {
+ None
+ }
+ }
+
+ override def get(startId: Option[Long], endId: Long): Array[(Long, T)] = {
+ val batchIds = fc.util().listStatus(metadataPath, batchFilesFilter)
+ .map(_.getPath.getName.toLong)
+ .filter { batchId =>
+ batchId <= endId && (startId.isEmpty || batchId >= startId.get)
+ }
+ batchIds.sorted.map(batchId => (batchId, get(batchId))).filter(_._2.isDefined).map {
+ case (batchId, metadataOption) =>
+ (batchId, metadataOption.get)
+ }
+ }
+
+ override def getLatest(): Option[(Long, T)] = {
+ val batchIds = fc.util().listStatus(metadataPath, batchFilesFilter)
+ .map(_.getPath.getName.toLong)
+ .sorted
+ .reverse
+ for (batchId <- batchIds) {
+ val batch = get(batchId)
+ if (batch.isDefined) {
+ return Some((batchId, batch.get))
+ }
+ }
+ None
+ }
+
+ override def stop(): Unit = {
--- End diff --
Get rid of stop? Seems like we would like to avoid relying on this method for correctness (since we need to handle abnormal termination). So I would just leave it out of the interface entirely.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13791][SQL]Add MetadataLog and HDFSMeta...
Posted by zsxwing <gi...@git.apache.org>.
Github user zsxwing commented on the pull request:
https://github.com/apache/spark/pull/11625#issuecomment-194615182
cc @marmbrus @tdas
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13791][SQL]Add MetadataLog and HDFSMeta...
Posted by zsxwing <gi...@git.apache.org>.
Github user zsxwing commented on a diff in the pull request:
https://github.com/apache/spark/pull/11625#discussion_r55741113
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/HDFSMetadataLog.scala ---
@@ -0,0 +1,168 @@
+/*
+* Licensed to the Apache Software Foundation (ASF) under one or more
+* contributor license agreements. See the NOTICE file distributed with
+* this work for additional information regarding copyright ownership.
+* The ASF licenses this file to You under the Apache License, Version 2.0
+* (the "License"); you may not use this file except in compliance with
+* the License. You may obtain a copy of the License at
+*
+* http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing, software
+* distributed under the License is distributed on an "AS IS" BASIS,
+* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+* See the License for the specific language governing permissions and
+* limitations under the License.
+*/
+
+package org.apache.spark.sql.execution.streaming
+
+import java.io.{DataOutput, EOFException, IOException}
+import java.nio.ByteBuffer
+
+import scala.reflect.ClassTag
+
+import org.apache.hadoop.fs.{Path, PathFilter}
+
+import org.apache.spark.SparkException
+import org.apache.spark.serializer.JavaSerializer
+import org.apache.spark.sql.SQLContext
+import org.apache.spark.util.Utils
+
+/**
+ * A [[MetadataLog]] implementation based on HDFS. [[HDFSMetadataLog]] uses the specified `path`
+ * as the metadata storage.
+ *
+ * There should only be one [[HDFSMetadataLog]] using `path` at the same time. [[HDFSMetadataLog]]
+ * uses a file ".lock" in the directory to make sure there is always only one user using `path`.
+ * When [[HDFSMetadataLog]] is created, it will create a ".lock" file. If this step fails,
+ * [[HDFSMetadataLog]] will throw an exception saying there is someone using the same directory.
+ * When [[HDFSMetadataLog]] is stopped, it will delete the ".lock" file.
+ *
+ * However, in extreme case, e.g., power outage, [[HDFSMetadataLog]] won't be able to delete ".lock"
+ * file. Then [[HDFSMetadataLog]] cannot be created even if no one is using the `path`. In such
+ * case, the user has to delete the lock file in the exception message to restart their application.
+ *
+ * Note: [[HDFSMetadataLog]] doesn't support S3-like file systems as they don't guarantee listing
+ * files in a directory always shows the latest files.
+ */
+class HDFSMetadataLog[T: ClassTag](sqlContext: SQLContext, path: String) extends MetadataLog[T] {
+
+ private val metadataPath = new Path(path)
+ private val fs = metadataPath.getFileSystem(sqlContext.sparkContext.hadoopConfiguration)
+ if (!fs.exists(metadataPath)) {
+ fs.mkdirs(metadataPath)
+ }
+
+ tryAcquireLock()
+
+ /**
+ * A `PathFilter` to filter only batch files
+ */
+ private val batchFilesFilter = new PathFilter {
+ override def accept(path: Path): Boolean = try {
+ path.getName.toLong
+ true
+ } catch {
+ case _: NumberFormatException => false
+ }
+ }
+
+ private val serializer = new JavaSerializer(sqlContext.sparkContext.conf).newInstance()
+
+ private def tryAcquireLock(): Unit = {
+ val lockFile = new Path(metadataPath, ".lock")
+ fs.deleteOnExit(lockFile)
+ try {
+ // Must set overwrite to false otherwise HDFS allows multiple clients to create the same file.
+ // When overwrite is false, the latter one will receive FileAlreadyExistsException
+ fs.create(lockFile, false).close()
+ } catch {
+ case e: IOException =>
+ throw new SparkException(s"There are another user using $path. Please check that. " +
+ s"If nobody is using it, please delete ${lockFile}", e)
+ }
+ }
+
+ private def batchFile(batchId: Long): Path = {
+ new Path(metadataPath, batchId.toString)
+ }
+
+ override def add(batchId: Long, metadata: T): Unit = {
+ get(batchId).getOrElse {
+ // Only write metadata when the batch has not yet been written.
+ val buffer = serializer.serialize(metadata)
+ val output = try {
+ fs.create(batchFile(batchId))
+ } catch {
+ case e: IOException if "java.lang.InterruptedException" == e.getMessage =>
+ // create may convert InterruptedException to IOException. Let's convert it back to
+ // InterruptedException so that this failure won't crash StreamExecution
+ throw new InterruptedException("Creating file is interrupted")
+ }
+ try {
+ output.writeInt(buffer.remaining())
+ Utils.writeByteBuffer(buffer, output: DataOutput)
--- End diff --
There are two methods in `Utils`:
```
def writeByteBuffer(bb: ByteBuffer, out: DataOutput): Unit
def writeByteBuffer(bb: ByteBuffer, out: OutputStream): Unit
```
`FSDataOutputStream` is both `DataOutput` and `OutputStream`. The compiler doesn't know to call which one. So I need to add the type here.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13791][SQL]Add MetadataLog and HDFSMeta...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11625#issuecomment-196517995
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/53103/
Test FAILed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13791][SQL]Add MetadataLog and HDFSMeta...
Posted by marmbrus <gi...@git.apache.org>.
Github user marmbrus commented on a diff in the pull request:
https://github.com/apache/spark/pull/11625#discussion_r55739913
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/HDFSMetadataLog.scala ---
@@ -0,0 +1,168 @@
+/*
+* Licensed to the Apache Software Foundation (ASF) under one or more
+* contributor license agreements. See the NOTICE file distributed with
+* this work for additional information regarding copyright ownership.
+* The ASF licenses this file to You under the Apache License, Version 2.0
+* (the "License"); you may not use this file except in compliance with
+* the License. You may obtain a copy of the License at
+*
+* http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing, software
+* distributed under the License is distributed on an "AS IS" BASIS,
+* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+* See the License for the specific language governing permissions and
+* limitations under the License.
+*/
+
+package org.apache.spark.sql.execution.streaming
+
+import java.io.{DataOutput, EOFException, IOException}
+import java.nio.ByteBuffer
+
+import scala.reflect.ClassTag
+
+import org.apache.hadoop.fs.{Path, PathFilter}
+
+import org.apache.spark.SparkException
+import org.apache.spark.serializer.JavaSerializer
+import org.apache.spark.sql.SQLContext
+import org.apache.spark.util.Utils
+
+/**
+ * A [[MetadataLog]] implementation based on HDFS. [[HDFSMetadataLog]] uses the specified `path`
+ * as the metadata storage.
+ *
+ * There should only be one [[HDFSMetadataLog]] using `path` at the same time. [[HDFSMetadataLog]]
+ * uses a file ".lock" in the directory to make sure there is always only one user using `path`.
+ * When [[HDFSMetadataLog]] is created, it will create a ".lock" file. If this step fails,
+ * [[HDFSMetadataLog]] will throw an exception saying there is someone using the same directory.
+ * When [[HDFSMetadataLog]] is stopped, it will delete the ".lock" file.
+ *
+ * However, in extreme case, e.g., power outage, [[HDFSMetadataLog]] won't be able to delete ".lock"
+ * file. Then [[HDFSMetadataLog]] cannot be created even if no one is using the `path`. In such
+ * case, the user has to delete the lock file in the exception message to restart their application.
+ *
+ * Note: [[HDFSMetadataLog]] doesn't support S3-like file systems as they don't guarantee listing
+ * files in a directory always shows the latest files.
+ */
+class HDFSMetadataLog[T: ClassTag](sqlContext: SQLContext, path: String) extends MetadataLog[T] {
+
+ private val metadataPath = new Path(path)
+ private val fs = metadataPath.getFileSystem(sqlContext.sparkContext.hadoopConfiguration)
+ if (!fs.exists(metadataPath)) {
+ fs.mkdirs(metadataPath)
+ }
+
+ tryAcquireLock()
+
+ /**
+ * A `PathFilter` to filter only batch files
+ */
+ private val batchFilesFilter = new PathFilter {
+ override def accept(path: Path): Boolean = try {
+ path.getName.toLong
+ true
+ } catch {
+ case _: NumberFormatException => false
+ }
+ }
+
+ private val serializer = new JavaSerializer(sqlContext.sparkContext.conf).newInstance()
+
+ private def tryAcquireLock(): Unit = {
--- End diff --
Seems like this abstraction would be more powerful if we threw a `ConcurrentUpdateException` and then the user could decide if that is okay. If all you are trying to get is idempotence (the file sink) then you can ignore it. If you are trying to do mutual exclusion (stream execution trying to define the offsets in each batch id) you can terminate the stream.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13791][SQL]Add MetadataLog and HDFSMeta...
Posted by zsxwing <gi...@git.apache.org>.
Github user zsxwing commented on a diff in the pull request:
https://github.com/apache/spark/pull/11625#discussion_r55742214
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/HDFSMetadataLog.scala ---
@@ -0,0 +1,168 @@
+/*
+* Licensed to the Apache Software Foundation (ASF) under one or more
+* contributor license agreements. See the NOTICE file distributed with
+* this work for additional information regarding copyright ownership.
+* The ASF licenses this file to You under the Apache License, Version 2.0
+* (the "License"); you may not use this file except in compliance with
+* the License. You may obtain a copy of the License at
+*
+* http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing, software
+* distributed under the License is distributed on an "AS IS" BASIS,
+* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+* See the License for the specific language governing permissions and
+* limitations under the License.
+*/
+
+package org.apache.spark.sql.execution.streaming
+
+import java.io.{DataOutput, EOFException, IOException}
+import java.nio.ByteBuffer
+
+import scala.reflect.ClassTag
+
+import org.apache.hadoop.fs.{Path, PathFilter}
+
+import org.apache.spark.SparkException
+import org.apache.spark.serializer.JavaSerializer
+import org.apache.spark.sql.SQLContext
+import org.apache.spark.util.Utils
+
+/**
+ * A [[MetadataLog]] implementation based on HDFS. [[HDFSMetadataLog]] uses the specified `path`
+ * as the metadata storage.
+ *
+ * There should only be one [[HDFSMetadataLog]] using `path` at the same time. [[HDFSMetadataLog]]
+ * uses a file ".lock" in the directory to make sure there is always only one user using `path`.
+ * When [[HDFSMetadataLog]] is created, it will create a ".lock" file. If this step fails,
+ * [[HDFSMetadataLog]] will throw an exception saying there is someone using the same directory.
+ * When [[HDFSMetadataLog]] is stopped, it will delete the ".lock" file.
+ *
+ * However, in extreme case, e.g., power outage, [[HDFSMetadataLog]] won't be able to delete ".lock"
+ * file. Then [[HDFSMetadataLog]] cannot be created even if no one is using the `path`. In such
+ * case, the user has to delete the lock file in the exception message to restart their application.
+ *
+ * Note: [[HDFSMetadataLog]] doesn't support S3-like file systems as they don't guarantee listing
+ * files in a directory always shows the latest files.
+ */
+class HDFSMetadataLog[T: ClassTag](sqlContext: SQLContext, path: String) extends MetadataLog[T] {
+
+ private val metadataPath = new Path(path)
+ private val fs = metadataPath.getFileSystem(sqlContext.sparkContext.hadoopConfiguration)
+ if (!fs.exists(metadataPath)) {
+ fs.mkdirs(metadataPath)
+ }
+
+ tryAcquireLock()
+
+ /**
+ * A `PathFilter` to filter only batch files
+ */
+ private val batchFilesFilter = new PathFilter {
+ override def accept(path: Path): Boolean = try {
+ path.getName.toLong
+ true
+ } catch {
+ case _: NumberFormatException => false
+ }
+ }
+
+ private val serializer = new JavaSerializer(sqlContext.sparkContext.conf).newInstance()
+
+ private def tryAcquireLock(): Unit = {
+ val lockFile = new Path(metadataPath, ".lock")
+ fs.deleteOnExit(lockFile)
+ try {
+ // Must set overwrite to false otherwise HDFS allows multiple clients to create the same file.
+ // When overwrite is false, the latter one will receive FileAlreadyExistsException
+ fs.create(lockFile, false).close()
+ } catch {
+ case e: IOException =>
+ throw new SparkException(s"There are another user using $path. Please check that. " +
+ s"If nobody is using it, please delete ${lockFile}", e)
+ }
+ }
+
+ private def batchFile(batchId: Long): Path = {
+ new Path(metadataPath, batchId.toString)
+ }
+
+ override def add(batchId: Long, metadata: T): Unit = {
+ get(batchId).getOrElse {
+ // Only write metadata when the batch has not yet been written.
+ val buffer = serializer.serialize(metadata)
+ val output = try {
+ fs.create(batchFile(batchId))
+ } catch {
+ case e: IOException if "java.lang.InterruptedException" == e.getMessage =>
--- End diff --
It's not in `cause`. See: https://github.com/apache/hadoop/blob/release-2.4.0/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/util/Shell.java#L508
Although this one has been fixed in [HADOOP-12074](https://issues.apache.org/jira/browse/HADOOP-12074), we still need to support the old versions.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13791][SQL]Add MetadataLog and HDFSMeta...
Posted by marmbrus <gi...@git.apache.org>.
Github user marmbrus commented on a diff in the pull request:
https://github.com/apache/spark/pull/11625#discussion_r55740192
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/HDFSMetadataLog.scala ---
@@ -0,0 +1,168 @@
+/*
+* Licensed to the Apache Software Foundation (ASF) under one or more
+* contributor license agreements. See the NOTICE file distributed with
+* this work for additional information regarding copyright ownership.
+* The ASF licenses this file to You under the Apache License, Version 2.0
+* (the "License"); you may not use this file except in compliance with
+* the License. You may obtain a copy of the License at
+*
+* http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing, software
+* distributed under the License is distributed on an "AS IS" BASIS,
+* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+* See the License for the specific language governing permissions and
+* limitations under the License.
+*/
+
+package org.apache.spark.sql.execution.streaming
+
+import java.io.{DataOutput, EOFException, IOException}
+import java.nio.ByteBuffer
+
+import scala.reflect.ClassTag
+
+import org.apache.hadoop.fs.{Path, PathFilter}
+
+import org.apache.spark.SparkException
+import org.apache.spark.serializer.JavaSerializer
+import org.apache.spark.sql.SQLContext
+import org.apache.spark.util.Utils
+
+/**
+ * A [[MetadataLog]] implementation based on HDFS. [[HDFSMetadataLog]] uses the specified `path`
+ * as the metadata storage.
+ *
+ * There should only be one [[HDFSMetadataLog]] using `path` at the same time. [[HDFSMetadataLog]]
+ * uses a file ".lock" in the directory to make sure there is always only one user using `path`.
+ * When [[HDFSMetadataLog]] is created, it will create a ".lock" file. If this step fails,
+ * [[HDFSMetadataLog]] will throw an exception saying there is someone using the same directory.
+ * When [[HDFSMetadataLog]] is stopped, it will delete the ".lock" file.
+ *
+ * However, in extreme case, e.g., power outage, [[HDFSMetadataLog]] won't be able to delete ".lock"
+ * file. Then [[HDFSMetadataLog]] cannot be created even if no one is using the `path`. In such
+ * case, the user has to delete the lock file in the exception message to restart their application.
+ *
+ * Note: [[HDFSMetadataLog]] doesn't support S3-like file systems as they don't guarantee listing
+ * files in a directory always shows the latest files.
+ */
+class HDFSMetadataLog[T: ClassTag](sqlContext: SQLContext, path: String) extends MetadataLog[T] {
+
+ private val metadataPath = new Path(path)
+ private val fs = metadataPath.getFileSystem(sqlContext.sparkContext.hadoopConfiguration)
+ if (!fs.exists(metadataPath)) {
+ fs.mkdirs(metadataPath)
+ }
+
+ tryAcquireLock()
+
+ /**
+ * A `PathFilter` to filter only batch files
+ */
+ private val batchFilesFilter = new PathFilter {
+ override def accept(path: Path): Boolean = try {
+ path.getName.toLong
+ true
+ } catch {
+ case _: NumberFormatException => false
+ }
+ }
+
+ private val serializer = new JavaSerializer(sqlContext.sparkContext.conf).newInstance()
+
+ private def tryAcquireLock(): Unit = {
--- End diff --
You could also return a boolean instead of throwing an exception.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13791][SQL]Add MetadataLog and HDFSMeta...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11625#issuecomment-195018692
Merged build finished. Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13791][SQL]Add MetadataLog and HDFSMeta...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11625#issuecomment-194613998
**[Test build #52801 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/52801/consoleFull)** for PR 11625 at commit [`c32cd96`](https://github.com/apache/spark/commit/c32cd963ba9bdd1ec0eec6b74bdbf361c00a2bf9).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13791][SQL]Add MetadataLog and HDFSMeta...
Posted by marmbrus <gi...@git.apache.org>.
Github user marmbrus commented on a diff in the pull request:
https://github.com/apache/spark/pull/11625#discussion_r55740102
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/HDFSMetadataLog.scala ---
@@ -0,0 +1,168 @@
+/*
+* Licensed to the Apache Software Foundation (ASF) under one or more
+* contributor license agreements. See the NOTICE file distributed with
+* this work for additional information regarding copyright ownership.
+* The ASF licenses this file to You under the Apache License, Version 2.0
+* (the "License"); you may not use this file except in compliance with
+* the License. You may obtain a copy of the License at
+*
+* http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing, software
+* distributed under the License is distributed on an "AS IS" BASIS,
+* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+* See the License for the specific language governing permissions and
+* limitations under the License.
+*/
+
+package org.apache.spark.sql.execution.streaming
+
+import java.io.{DataOutput, EOFException, IOException}
+import java.nio.ByteBuffer
+
+import scala.reflect.ClassTag
+
+import org.apache.hadoop.fs.{Path, PathFilter}
+
+import org.apache.spark.SparkException
+import org.apache.spark.serializer.JavaSerializer
+import org.apache.spark.sql.SQLContext
+import org.apache.spark.util.Utils
+
+/**
+ * A [[MetadataLog]] implementation based on HDFS. [[HDFSMetadataLog]] uses the specified `path`
+ * as the metadata storage.
+ *
+ * There should only be one [[HDFSMetadataLog]] using `path` at the same time. [[HDFSMetadataLog]]
+ * uses a file ".lock" in the directory to make sure there is always only one user using `path`.
+ * When [[HDFSMetadataLog]] is created, it will create a ".lock" file. If this step fails,
+ * [[HDFSMetadataLog]] will throw an exception saying there is someone using the same directory.
+ * When [[HDFSMetadataLog]] is stopped, it will delete the ".lock" file.
+ *
+ * However, in extreme case, e.g., power outage, [[HDFSMetadataLog]] won't be able to delete ".lock"
+ * file. Then [[HDFSMetadataLog]] cannot be created even if no one is using the `path`. In such
+ * case, the user has to delete the lock file in the exception message to restart their application.
+ *
+ * Note: [[HDFSMetadataLog]] doesn't support S3-like file systems as they don't guarantee listing
+ * files in a directory always shows the latest files.
+ */
+class HDFSMetadataLog[T: ClassTag](sqlContext: SQLContext, path: String) extends MetadataLog[T] {
+
+ private val metadataPath = new Path(path)
+ private val fs = metadataPath.getFileSystem(sqlContext.sparkContext.hadoopConfiguration)
+ if (!fs.exists(metadataPath)) {
+ fs.mkdirs(metadataPath)
+ }
+
+ tryAcquireLock()
+
+ /**
+ * A `PathFilter` to filter only batch files
+ */
+ private val batchFilesFilter = new PathFilter {
+ override def accept(path: Path): Boolean = try {
+ path.getName.toLong
+ true
+ } catch {
+ case _: NumberFormatException => false
+ }
+ }
+
+ private val serializer = new JavaSerializer(sqlContext.sparkContext.conf).newInstance()
+
+ private def tryAcquireLock(): Unit = {
--- End diff --
Basically, I don't believe that stale locks are only going to occur in extreme circumstances. If the JVM locks due to an OOM or if the container spark is running in is killed or if you kill -9. All of these become unrecoverable.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13791][SQL]Add MetadataLog and HDFSMeta...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11625#issuecomment-196517933
**[Test build #53103 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/53103/consoleFull)** for PR 11625 at commit [`4c27c6e`](https://github.com/apache/spark/commit/4c27c6eb2ba17cebacee3a77b2ce5643e6da0329).
* This patch **fails MiMa tests**.
* This patch merges cleanly.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13791][SQL]Add MetadataLog and HDFSMeta...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11625#issuecomment-196600585
**[Test build #53137 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/53137/consoleFull)** for PR 11625 at commit [`1c82e56`](https://github.com/apache/spark/commit/1c82e5681c35e1a169a9e3f49720adaa19462cce).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13791][SQL]Add MetadataLog and HDFSMeta...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11625#issuecomment-194641288
Merged build finished. Test FAILed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13791][SQL]Add MetadataLog and HDFSMeta...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11625#issuecomment-194641291
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/52801/
Test FAILed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13791][SQL]Add MetadataLog and HDFSMeta...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11625#issuecomment-196517993
Merged build finished. Test FAILed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13791][SQL]Add MetadataLog and HDFSMeta...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11625#issuecomment-196530006
**[Test build #53112 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/53112/consoleFull)** for PR 11625 at commit [`4c27c6e`](https://github.com/apache/spark/commit/4c27c6eb2ba17cebacee3a77b2ce5643e6da0329).
* This patch **fails MiMa tests**.
* This patch merges cleanly.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13791][SQL]Add MetadataLog and HDFSMeta...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11625#issuecomment-195018699
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/52835/
Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13791][SQL]Add MetadataLog and HDFSMeta...
Posted by marmbrus <gi...@git.apache.org>.
Github user marmbrus commented on the pull request:
https://github.com/apache/spark/pull/11625#issuecomment-196620147
Thanks! Merging to master.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13791][SQL]Add MetadataLog and HDFSMeta...
Posted by marmbrus <gi...@git.apache.org>.
Github user marmbrus commented on a diff in the pull request:
https://github.com/apache/spark/pull/11625#discussion_r56094406
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/MetadataLog.scala ---
@@ -0,0 +1,56 @@
+/*
+* Licensed to the Apache Software Foundation (ASF) under one or more
+* contributor license agreements. See the NOTICE file distributed with
+* this work for additional information regarding copyright ownership.
+* The ASF licenses this file to You under the Apache License, Version 2.0
+* (the "License"); you may not use this file except in compliance with
+* the License. You may obtain a copy of the License at
+*
+* http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing, software
+* distributed under the License is distributed on an "AS IS" BASIS,
+* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+* See the License for the specific language governing permissions and
+* limitations under the License.
+*/
+
+package org.apache.spark.sql.execution.streaming
+
+/**
+ * A general MetadataLog that supports the following features:
+ *
+ * - Allow the user to store a metadata object for each batch.
+ * - Allow the user to query the latest batch id.
+ * - Allow the user to query the metadata object of a specified batch id.
+ * - Allow the user to query metadata objects in a range of batch ids.
+ */
+trait MetadataLog[T] {
+
+ /**
+ * Store the metadata for the specified batchId. If batchId's metadata has already been stored,
+ * this method does nothing.
--- End diff --
Update these docs
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13791][SQL]Add MetadataLog and HDFSMeta...
Posted by marmbrus <gi...@git.apache.org>.
Github user marmbrus commented on a diff in the pull request:
https://github.com/apache/spark/pull/11625#discussion_r55740501
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/HDFSMetadataLog.scala ---
@@ -0,0 +1,168 @@
+/*
+* Licensed to the Apache Software Foundation (ASF) under one or more
+* contributor license agreements. See the NOTICE file distributed with
+* this work for additional information regarding copyright ownership.
+* The ASF licenses this file to You under the Apache License, Version 2.0
+* (the "License"); you may not use this file except in compliance with
+* the License. You may obtain a copy of the License at
+*
+* http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing, software
+* distributed under the License is distributed on an "AS IS" BASIS,
+* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+* See the License for the specific language governing permissions and
+* limitations under the License.
+*/
+
+package org.apache.spark.sql.execution.streaming
+
+import java.io.{DataOutput, EOFException, IOException}
+import java.nio.ByteBuffer
+
+import scala.reflect.ClassTag
+
+import org.apache.hadoop.fs.{Path, PathFilter}
+
+import org.apache.spark.SparkException
+import org.apache.spark.serializer.JavaSerializer
+import org.apache.spark.sql.SQLContext
+import org.apache.spark.util.Utils
+
+/**
+ * A [[MetadataLog]] implementation based on HDFS. [[HDFSMetadataLog]] uses the specified `path`
+ * as the metadata storage.
+ *
+ * There should only be one [[HDFSMetadataLog]] using `path` at the same time. [[HDFSMetadataLog]]
+ * uses a file ".lock" in the directory to make sure there is always only one user using `path`.
+ * When [[HDFSMetadataLog]] is created, it will create a ".lock" file. If this step fails,
+ * [[HDFSMetadataLog]] will throw an exception saying there is someone using the same directory.
+ * When [[HDFSMetadataLog]] is stopped, it will delete the ".lock" file.
+ *
+ * However, in extreme case, e.g., power outage, [[HDFSMetadataLog]] won't be able to delete ".lock"
+ * file. Then [[HDFSMetadataLog]] cannot be created even if no one is using the `path`. In such
+ * case, the user has to delete the lock file in the exception message to restart their application.
+ *
+ * Note: [[HDFSMetadataLog]] doesn't support S3-like file systems as they don't guarantee listing
+ * files in a directory always shows the latest files.
+ */
+class HDFSMetadataLog[T: ClassTag](sqlContext: SQLContext, path: String) extends MetadataLog[T] {
+
+ private val metadataPath = new Path(path)
+ private val fs = metadataPath.getFileSystem(sqlContext.sparkContext.hadoopConfiguration)
+ if (!fs.exists(metadataPath)) {
+ fs.mkdirs(metadataPath)
+ }
+
+ tryAcquireLock()
+
+ /**
+ * A `PathFilter` to filter only batch files
+ */
+ private val batchFilesFilter = new PathFilter {
+ override def accept(path: Path): Boolean = try {
+ path.getName.toLong
+ true
+ } catch {
+ case _: NumberFormatException => false
+ }
+ }
+
+ private val serializer = new JavaSerializer(sqlContext.sparkContext.conf).newInstance()
+
+ private def tryAcquireLock(): Unit = {
+ val lockFile = new Path(metadataPath, ".lock")
+ fs.deleteOnExit(lockFile)
+ try {
+ // Must set overwrite to false otherwise HDFS allows multiple clients to create the same file.
+ // When overwrite is false, the latter one will receive FileAlreadyExistsException
+ fs.create(lockFile, false).close()
+ } catch {
+ case e: IOException =>
+ throw new SparkException(s"There are another user using $path. Please check that. " +
+ s"If nobody is using it, please delete ${lockFile}", e)
+ }
+ }
+
+ private def batchFile(batchId: Long): Path = {
+ new Path(metadataPath, batchId.toString)
+ }
+
+ override def add(batchId: Long, metadata: T): Unit = {
+ get(batchId).getOrElse {
+ // Only write metadata when the batch has not yet been written.
+ val buffer = serializer.serialize(metadata)
+ val output = try {
+ fs.create(batchFile(batchId))
+ } catch {
+ case e: IOException if "java.lang.InterruptedException" == e.getMessage =>
+ // create may convert InterruptedException to IOException. Let's convert it back to
+ // InterruptedException so that this failure won't crash StreamExecution
+ throw new InterruptedException("Creating file is interrupted")
+ }
+ try {
+ output.writeInt(buffer.remaining())
+ Utils.writeByteBuffer(buffer, output: DataOutput)
--- End diff --
Why the type ascription?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13791][SQL]Add MetadataLog and HDFSMeta...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11625#issuecomment-196483089
Merged build finished. Test FAILed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13791][SQL]Add MetadataLog and HDFSMeta...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11625#issuecomment-195018009
**[Test build #52835 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/52835/consoleFull)** for PR 11625 at commit [`d77bf39`](https://github.com/apache/spark/commit/d77bf392a958a75d5436a1ca6921a254a87e9628).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13791][SQL]Add MetadataLog and HDFSMeta...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11625#issuecomment-194979104
**[Test build #52835 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/52835/consoleFull)** for PR 11625 at commit [`d77bf39`](https://github.com/apache/spark/commit/d77bf392a958a75d5436a1ca6921a254a87e9628).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13791][SQL]Add MetadataLog and HDFSMeta...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11625#issuecomment-196436690
**[Test build #53083 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/53083/consoleFull)** for PR 11625 at commit [`7a52adc`](https://github.com/apache/spark/commit/7a52adc35d7e7100bc9271d54196b7eb85f5015b).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13791][SQL]Add MetadataLog and HDFSMeta...
Posted by zsxwing <gi...@git.apache.org>.
Github user zsxwing commented on a diff in the pull request:
https://github.com/apache/spark/pull/11625#discussion_r55623001
--- Diff: sql/core/src/test/scala/org/apache/spark/sql/streaming/FileStreamSourceSuite.scala ---
@@ -360,59 +357,6 @@ class FileStreamSourceSuite extends FileStreamSourceTest with SharedSQLContext {
Utils.deleteRecursively(tmp)
}
- test("fault tolerance with corrupted metadata file") {
--- End diff --
Removed these tests as they were testing the old metadata file which has been removed in this PR
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13791][SQL]Add MetadataLog and HDFSMeta...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11625#issuecomment-196530047
Merged build finished. Test FAILed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13791][SQL]Add MetadataLog and HDFSMeta...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11625#issuecomment-196483091
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/53083/
Test FAILed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13791][SQL]Add MetadataLog and HDFSMeta...
Posted by zsxwing <gi...@git.apache.org>.
Github user zsxwing commented on a diff in the pull request:
https://github.com/apache/spark/pull/11625#discussion_r55775160
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/HDFSMetadataLog.scala ---
@@ -0,0 +1,168 @@
+/*
+* Licensed to the Apache Software Foundation (ASF) under one or more
+* contributor license agreements. See the NOTICE file distributed with
+* this work for additional information regarding copyright ownership.
+* The ASF licenses this file to You under the Apache License, Version 2.0
+* (the "License"); you may not use this file except in compliance with
+* the License. You may obtain a copy of the License at
+*
+* http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing, software
+* distributed under the License is distributed on an "AS IS" BASIS,
+* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+* See the License for the specific language governing permissions and
+* limitations under the License.
+*/
+
+package org.apache.spark.sql.execution.streaming
+
+import java.io.{DataOutput, EOFException, IOException}
+import java.nio.ByteBuffer
+
+import scala.reflect.ClassTag
+
+import org.apache.hadoop.fs.{Path, PathFilter}
+
+import org.apache.spark.SparkException
+import org.apache.spark.serializer.JavaSerializer
+import org.apache.spark.sql.SQLContext
+import org.apache.spark.util.Utils
+
+/**
+ * A [[MetadataLog]] implementation based on HDFS. [[HDFSMetadataLog]] uses the specified `path`
+ * as the metadata storage.
+ *
+ * There should only be one [[HDFSMetadataLog]] using `path` at the same time. [[HDFSMetadataLog]]
+ * uses a file ".lock" in the directory to make sure there is always only one user using `path`.
+ * When [[HDFSMetadataLog]] is created, it will create a ".lock" file. If this step fails,
+ * [[HDFSMetadataLog]] will throw an exception saying there is someone using the same directory.
+ * When [[HDFSMetadataLog]] is stopped, it will delete the ".lock" file.
+ *
+ * However, in extreme case, e.g., power outage, [[HDFSMetadataLog]] won't be able to delete ".lock"
+ * file. Then [[HDFSMetadataLog]] cannot be created even if no one is using the `path`. In such
+ * case, the user has to delete the lock file in the exception message to restart their application.
+ *
+ * Note: [[HDFSMetadataLog]] doesn't support S3-like file systems as they don't guarantee listing
+ * files in a directory always shows the latest files.
+ */
+class HDFSMetadataLog[T: ClassTag](sqlContext: SQLContext, path: String) extends MetadataLog[T] {
+
+ private val metadataPath = new Path(path)
+ private val fs = metadataPath.getFileSystem(sqlContext.sparkContext.hadoopConfiguration)
+ if (!fs.exists(metadataPath)) {
+ fs.mkdirs(metadataPath)
+ }
+
+ tryAcquireLock()
+
+ /**
+ * A `PathFilter` to filter only batch files
+ */
+ private val batchFilesFilter = new PathFilter {
+ override def accept(path: Path): Boolean = try {
+ path.getName.toLong
+ true
+ } catch {
+ case _: NumberFormatException => false
+ }
+ }
+
+ private val serializer = new JavaSerializer(sqlContext.sparkContext.conf).newInstance()
+
+ private def tryAcquireLock(): Unit = {
--- End diff --
If we don't use a global `.lock` file, there are two cases when writing a log entry fails because of FileAlreadyExistsException:
1. There is another HDFSMetadataLog using the same path
2. The file is corrupted. We just restarted from a failure and tried to rerun a batch.
For case 1, we want to throw `ConcurrentUpdateException`; for case 2, we need to override the file.
As we need to figure out which situation we are in, we will try to read the file to see if it's completed. However, we also have two possibilities if we find the file is corrupted:
1. Another HDFSMetadataLog is writing the file.
2. Nobody is using the path and the file is corrupted.
So how can we know which case is right?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-13791][SQL]Add MetadataLog and HDFSMeta...
Posted by zsxwing <gi...@git.apache.org>.
Github user zsxwing commented on the pull request:
https://github.com/apache/spark/pull/11625#issuecomment-196524485
retest this please
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org