[GitHub] spark pull request #20906: [SPARK-23561][SS] Pull continuous processing out ...
Github user jose-torres closed the pull request at: https://github.com/apache/spark/pull/20906 --- - To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20906: [SPARK-23561][SS] Pull continuous processing out ...
Github user cloud-fan commented on a diff in the pull request:
https://github.com/apache/spark/pull/20906#discussion_r179713596
--- Diff:
sql/core/src/main/java/org/apache/spark/sql/sources/v2/writer/streaming/StreamWriter.java
---
@@ -29,14 +31,15 @@
* increasing numeric ID. This writer handles commits and aborts for each
successive epoch.
*/
@InterfaceStability.Evolving
-public interface StreamWriter extends DataSourceWriter {
--- End diff --
We should remove `extend DataSourceWriter` in these mix-in interfaces, so
that it can work for both `DataSourceWriter` and `StreamWriter`.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20906: [SPARK-23561][SS] Pull continuous processing out ...
Github user cloud-fan commented on a diff in the pull request:
https://github.com/apache/spark/pull/20906#discussion_r179685434
--- Diff:
sql/core/src/main/java/org/apache/spark/sql/sources/v2/writer/streaming/StreamWriter.java
---
@@ -29,14 +31,15 @@
* increasing numeric ID. This writer handles commits and aborts for each
successive epoch.
*/
@InterfaceStability.Evolving
-public interface StreamWriter extends DataSourceWriter {
--- End diff --
By design we would create more mix-in interfaces for `DataSourceWriter`,
like `SupportsWriteColumnarBatch`. If the `StreamWriter` doesn't extend
`DataSourceWriter`, how shall we apply those mix-in interfaces for
`StreamWriter`?
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20906: [SPARK-23561][SS] Pull continuous processing out ...
Github user jose-torres commented on a diff in the pull request:
https://github.com/apache/spark/pull/20906#discussion_r177288972
--- Diff:
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/continuous/ContinuousWriteExec.scala
---
@@ -0,0 +1,122 @@
+/*
+ * 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.continuous
+
+import scala.util.control.NonFatal
+
+import org.apache.spark.{SparkEnv, SparkException, TaskContext}
+import org.apache.spark.internal.Logging
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.Attribute
+import org.apache.spark.sql.execution.SparkPlan
+import
org.apache.spark.sql.execution.datasources.v2.{DataWritingSparkTask,
InternalRowDataWriterFactory}
+import
org.apache.spark.sql.execution.datasources.v2.DataWritingSparkTask.{logError,
logInfo}
+import org.apache.spark.sql.execution.streaming.StreamExecution
+import org.apache.spark.sql.sources.v2.writer._
+import org.apache.spark.sql.sources.v2.writer.streaming.StreamWriter
+import org.apache.spark.util.Utils
+
+case class ContinuousWriteExec(writer: StreamWriter, query: SparkPlan)
--- End diff --
Sure, works for me.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20906: [SPARK-23561][SS] Pull continuous processing out ...
Github user tdas commented on a diff in the pull request:
https://github.com/apache/spark/pull/20906#discussion_r177245022
--- Diff:
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/continuous/ContinuousWriteExec.scala
---
@@ -0,0 +1,122 @@
+/*
+ * 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.continuous
+
+import scala.util.control.NonFatal
+
+import org.apache.spark.{SparkEnv, SparkException, TaskContext}
+import org.apache.spark.internal.Logging
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.Attribute
+import org.apache.spark.sql.execution.SparkPlan
+import
org.apache.spark.sql.execution.datasources.v2.{DataWritingSparkTask,
InternalRowDataWriterFactory}
+import
org.apache.spark.sql.execution.datasources.v2.DataWritingSparkTask.{logError,
logInfo}
+import org.apache.spark.sql.execution.streaming.StreamExecution
+import org.apache.spark.sql.sources.v2.writer._
+import org.apache.spark.sql.sources.v2.writer.streaming.StreamWriter
+import org.apache.spark.util.Utils
+
+case class ContinuousWriteExec(writer: StreamWriter, query: SparkPlan)
--- End diff --
Isnt it better to keep the name consistent with WriterToDataSource? Say,
WriteToContinuousDataSourceExec?
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20906: [SPARK-23561][SS] Pull continuous processing out ...
Github user tdas commented on a diff in the pull request:
https://github.com/apache/spark/pull/20906#discussion_r177244722
--- Diff:
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/continuous/ContinuousWriteExec.scala
---
@@ -0,0 +1,122 @@
+/*
+ * 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.continuous
+
+import scala.util.control.NonFatal
+
+import org.apache.spark.{SparkEnv, SparkException, TaskContext}
+import org.apache.spark.internal.Logging
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.Attribute
+import org.apache.spark.sql.execution.SparkPlan
+import
org.apache.spark.sql.execution.datasources.v2.{DataWritingSparkTask,
InternalRowDataWriterFactory}
+import
org.apache.spark.sql.execution.datasources.v2.DataWritingSparkTask.{logError,
logInfo}
+import org.apache.spark.sql.execution.streaming.StreamExecution
+import org.apache.spark.sql.sources.v2.writer._
+import org.apache.spark.sql.sources.v2.writer.streaming.StreamWriter
+import org.apache.spark.util.Utils
+
+case class ContinuousWriteExec(writer: StreamWriter, query: SparkPlan)
+extends SparkPlan with Logging {
+ override def children: Seq[SparkPlan] = Seq(query)
+ override def output: Seq[Attribute] = Nil
+
+ override protected def doExecute(): RDD[InternalRow] = {
+val writerFactory = writer match {
+ case w: SupportsWriteInternalRow =>
w.createInternalRowWriterFactory()
+ case _ => new
InternalRowDataWriterFactory(writer.createWriterFactory(), query.schema)
+}
+
+val rdd = query.execute()
+val messages = new Array[WriterCommitMessage](rdd.partitions.length)
+
+logInfo(s"Start processing data source writer: $writer. " +
+ s"The input RDD has ${messages.length} partitions.")
+EpochCoordinatorRef.get(
--- End diff --
nit: Add comment on what this does.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20906: [SPARK-23561][SS] Pull continuous processing out ...
Github user tdas commented on a diff in the pull request:
https://github.com/apache/spark/pull/20906#discussion_r177275489
--- Diff:
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/continuous/ContinuousWriteExec.scala
---
@@ -0,0 +1,122 @@
+/*
+ * 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.continuous
+
+import scala.util.control.NonFatal
+
+import org.apache.spark.{SparkEnv, SparkException, TaskContext}
+import org.apache.spark.internal.Logging
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.Attribute
+import org.apache.spark.sql.execution.SparkPlan
+import
org.apache.spark.sql.execution.datasources.v2.{DataWritingSparkTask,
InternalRowDataWriterFactory}
+import
org.apache.spark.sql.execution.datasources.v2.DataWritingSparkTask.{logError,
logInfo}
+import org.apache.spark.sql.execution.streaming.StreamExecution
+import org.apache.spark.sql.sources.v2.writer._
+import org.apache.spark.sql.sources.v2.writer.streaming.StreamWriter
+import org.apache.spark.util.Utils
+
+case class ContinuousWriteExec(writer: StreamWriter, query: SparkPlan)
+extends SparkPlan with Logging {
+ override def children: Seq[SparkPlan] = Seq(query)
+ override def output: Seq[Attribute] = Nil
+
+ override protected def doExecute(): RDD[InternalRow] = {
+val writerFactory = writer match {
+ case w: SupportsWriteInternalRow =>
w.createInternalRowWriterFactory()
+ case _ => new
InternalRowDataWriterFactory(writer.createWriterFactory(), query.schema)
+}
+
+val rdd = query.execute()
+val messages = new Array[WriterCommitMessage](rdd.partitions.length)
+
+logInfo(s"Start processing data source writer: $writer. " +
+ s"The input RDD has ${messages.length} partitions.")
+EpochCoordinatorRef.get(
+
sparkContext.getLocalProperty(ContinuousExecution.EPOCH_COORDINATOR_ID_KEY),
+ sparkContext.env)
+ .askSync[Unit](SetWriterPartitions(rdd.getNumPartitions))
--- End diff --
nit: this indentation looks weird.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20906: [SPARK-23561][SS] Pull continuous processing out ...
Github user tdas commented on a diff in the pull request:
https://github.com/apache/spark/pull/20906#discussion_r177243246
--- Diff:
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/continuous/ContinuousWriteExec.scala
---
@@ -0,0 +1,122 @@
+/*
+ * 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.continuous
+
+import scala.util.control.NonFatal
+
+import org.apache.spark.{SparkEnv, SparkException, TaskContext}
+import org.apache.spark.internal.Logging
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.Attribute
+import org.apache.spark.sql.execution.SparkPlan
+import
org.apache.spark.sql.execution.datasources.v2.{DataWritingSparkTask,
InternalRowDataWriterFactory}
+import
org.apache.spark.sql.execution.datasources.v2.DataWritingSparkTask.{logError,
logInfo}
+import org.apache.spark.sql.execution.streaming.StreamExecution
+import org.apache.spark.sql.sources.v2.writer._
+import org.apache.spark.sql.sources.v2.writer.streaming.StreamWriter
+import org.apache.spark.util.Utils
+
+case class ContinuousWriteExec(writer: StreamWriter, query: SparkPlan)
--- End diff --
add docs
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20906: [SPARK-23561][SS] Pull continuous processing out ...
GitHub user jose-torres opened a pull request: https://github.com/apache/spark/pull/20906 [SPARK-23561][SS] Pull continuous processing out of WriteToDataSourceV2Exec. ## What changes were proposed in this pull request? Pull continuous processing out of WriteToDataSourceV2Exec into its own plan node ContinuousWriteExec. This is part of the work required to let StreamWriter stop inheriting from DataSourceWriter. ## How was this patch tested? existing tests - refactoring only You can merge this pull request into a Git repository by running: $ git pull https://github.com/jose-torres/spark SPARK-23561 Alternatively you can review and apply these changes as the patch at: https://github.com/apache/spark/pull/20906.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 #20906 commit 0e8d80da1350b950fd690ff7c762d07d0767eafd Author: Jose Torres Date: 2018-03-21T22:05:31Z partial commit 270f8ffe062d76e44a726753afc07c78348c3cc6 Author: Jose Torres Date: 2018-03-22T06:08:52Z make ContinuousWriteExec work commit 0cfeaeb6c0e6a3b500341852db8c53359120a753 Author: Jose Torres Date: 2018-03-23T02:44:10Z fix docs commit 7c375339bac0704c99ba6d87ee671dc3b7c0f531 Author: Jose Torres Date: 2018-03-26T16:47:16Z remove old path --- - To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org
