You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by tdas <gi...@git.apache.org> on 2016/06/13 23:36:00 UTC
[GitHub] spark pull request #13653: [SPARK-15933][SQL][STREAMING] Refactored DF reade...
GitHub user tdas opened a pull request:
https://github.com/apache/spark/pull/13653
[SPARK-15933][SQL][STREAMING] Refactored DF reader-writer to use readStream and writeStream for streaming DFs
## What changes were proposed in this pull request?
Currently, the DataFrameReader/Writer has method that are needed for streaming and non-streaming DFs. This is quite awkward because each method in them through runtime exception for one case or the other. So rather having half the methods throw runtime exceptions, its just better to have a different reader/writer API for streams.
## How was this patch tested?
Existing unit tests + two sets of unit tests for DataFrameReader/Writer and DataStreamReader/Writer.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/tdas/spark SPARK-15933
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/spark/pull/13653.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 #13653
----
----
---
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 issue #13653: [SPARK-15933][SQL][STREAMING] Refactored DF reader-write...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/13653
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/60452/
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 #13653: [SPARK-15933][SQL][STREAMING] Refactored DF reade...
Posted by zsxwing <gi...@git.apache.org>.
Github user zsxwing commented on a diff in the pull request:
https://github.com/apache/spark/pull/13653#discussion_r67077331
--- Diff: python/pyspark/sql/readwriter.py ---
@@ -905,6 +764,503 @@ def jdbc(self, url, table, mode=None, properties=None):
self._jwrite.mode(mode).jdbc(url, table, jprop)
+class DataStreamReader(object):
--- End diff --
nit: add new classes to `__all__ = ["DataFrameReader", "DataFrameWriter"]`
---
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 issue #13653: [SPARK-15933][SQL][STREAMING] Refactored DF reader-write...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/13653
**[Test build #60452 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/60452/consoleFull)** for PR 13653 at commit [`a59498b`](https://github.com/apache/spark/commit/a59498bbe86609bc206de9b229052f35071049cb).
* This patch **fails PySpark 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 issue #13653: [SPARK-15933][SQL][STREAMING] Refactored DF reader-write...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/13653
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/60460/
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 issue #13653: [SPARK-15933][SQL][STREAMING] Refactored DF reader-write...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/13653
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 #13653: [SPARK-15933][SQL][STREAMING] Refactored DF reade...
Posted by tdas <gi...@git.apache.org>.
Github user tdas commented on a diff in the pull request:
https://github.com/apache/spark/pull/13653#discussion_r66887779
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/streaming/DataStreamReader.scala ---
@@ -0,0 +1,288 @@
+/*
+ * 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.streaming
+
+import scala.collection.JavaConverters._
+
+import org.apache.spark.annotation.Experimental
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.{DataFrame, Dataset, SparkSession}
+import org.apache.spark.sql.execution.datasources.DataSource
+import org.apache.spark.sql.execution.streaming.StreamingRelation
+import org.apache.spark.sql.types.StructType
+
+@Experimental
--- End diff --
add 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 #13653: [SPARK-15933][SQL][STREAMING] Refactored DF reade...
Posted by tdas <gi...@git.apache.org>.
Github user tdas commented on a diff in the pull request:
https://github.com/apache/spark/pull/13653#discussion_r66888413
--- Diff: sql/core/src/test/scala/org/apache/spark/sql/streaming/test/DataStreamReaderWriterSuite.scala ---
@@ -371,76 +384,12 @@ class DataFrameReaderWriterSuite extends StreamTest with BeforeAndAfter {
private def newTextInput = Utils.createTempDir(namePrefix = "text").getCanonicalPath
- test("check trigger() can only be called on continuous queries") {
--- End diff --
Most of these tests are to check whether one method is called on the wrong type of DF or not. So all of these can be removed.
---
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 issue #13653: [SPARK-15933][SQL][STREAMING] Refactored DF reader-write...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/13653
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 issue #13653: [SPARK-15933][SQL][STREAMING] Refactored DF reader-write...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/13653
**[Test build #60516 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/60516/consoleFull)** for PR 13653 at commit [`536c25e`](https://github.com/apache/spark/commit/536c25e00b154a467279ab0055ae98002bd70847).
---
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 issue #13653: [SPARK-15933][SQL][STREAMING] Refactored DF reader-write...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/13653
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/60516/
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 #13653: [SPARK-15933][SQL][STREAMING] Refactored DF reade...
Posted by zsxwing <gi...@git.apache.org>.
Github user zsxwing commented on a diff in the pull request:
https://github.com/apache/spark/pull/13653#discussion_r67078670
--- Diff: sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingAggregationSuite.scala ---
@@ -40,6 +40,8 @@ class StreamingAggregationSuite extends StreamTest with BeforeAndAfterAll {
import testImplicits._
+
--- End diff --
nit: redundant spaces.
---
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 issue #13653: [SPARK-15933][SQL][STREAMING] Refactored DF reader-write...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/13653
**[Test build #60460 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/60460/consoleFull)** for PR 13653 at commit [`e118631`](https://github.com/apache/spark/commit/e118631fd5bbea962932a812addae21c89f6f4f2).
* This patch **fails Python style tests**.
* This patch merges cleanly.
* This patch adds the following public classes _(experimental)_:
* `class DataStreamReader(object):`
* `class DataStreamWriter(object):`
---
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 #13653: [SPARK-15933][SQL][STREAMING] Refactored DF reade...
Posted by marmbrus <gi...@git.apache.org>.
Github user marmbrus commented on a diff in the pull request:
https://github.com/apache/spark/pull/13653#discussion_r67023372
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/streaming/DataStreamWriter.scala ---
@@ -0,0 +1,401 @@
+/*
+ * 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.streaming
+
+import scala.collection.JavaConverters._
+
+import org.apache.spark.annotation.Experimental
+import org.apache.spark.sql.{AnalysisException, DataFrame, Dataset, ForeachWriter}
+import org.apache.spark.sql.execution.datasources.DataSource
+import org.apache.spark.sql.execution.streaming.{ForeachSink, MemoryPlan, MemorySink}
+
+/**
+ * :: Experimental ::
+ * Interface used to write a streaming [[Dataset]] to external storage systems (e.g. file systems,
+ * key-value stores, etc). Use [[Dataset.write]] to access this.
+ *
+ * @since 2.0.0
+ */
+@Experimental
+final class DataStreamWriter[T] private[sql](ds: Dataset[T]) {
+
+ private val df = ds.toDF()
+
+ /**
+ * :: Experimental ::
+ * Specifies how data of a streaming DataFrame/Dataset is written to a streaming sink.
+ * - `OutputMode.Append()`: only the new rows in the streaming DataFrame/Dataset will be
+ * written to the sink
+ * - `OutputMode.Complete()`: all the rows in the streaming DataFrame/Dataset will be written
+ * to the sink every time these is some updates
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def outputMode(outputMode: OutputMode): DataStreamWriter[T] = {
+ this.outputMode = outputMode
+ this
+ }
+
+
+ /**
+ * :: Experimental ::
+ * Specifies how data of a streaming DataFrame/Dataset is written to a streaming sink.
+ * - `append`: only the new rows in the streaming DataFrame/Dataset will be written to
+ * the sink
+ * - `complete`: all the rows in the streaming DataFrame/Dataset will be written to the sink
+ * every time these is some updates
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def outputMode(outputMode: String): DataStreamWriter[T] = {
+ this.outputMode = outputMode.toLowerCase match {
+ case "append" =>
+ OutputMode.Append
+ case "complete" =>
+ OutputMode.Complete
+ case _ =>
+ throw new IllegalArgumentException(s"Unknown output mode $outputMode. " +
+ "Accepted output modes are 'append' and 'complete'")
+ }
+ this
+ }
+
+ /**
+ * :: Experimental ::
+ * Set the trigger for the stream query. The default value is `ProcessingTime(0)` and it will run
+ * the query as fast as possible.
+ *
+ * Scala Example:
+ * {{{
+ * df.write.trigger(ProcessingTime("10 seconds"))
+ *
+ * import scala.concurrent.duration._
+ * df.write.trigger(ProcessingTime(10.seconds))
+ * }}}
+ *
+ * Java Example:
+ * {{{
+ * df.write.trigger(ProcessingTime.create("10 seconds"))
+ *
+ * import java.util.concurrent.TimeUnit
+ * df.write.trigger(ProcessingTime.create(10, TimeUnit.SECONDS))
+ * }}}
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def trigger(trigger: Trigger): DataStreamWriter[T] = {
+ this.trigger = trigger
+ this
+ }
+
+
+ /**
+ * :: Experimental ::
+ * Specifies the name of the [[ContinuousQuery]] that can be started with `startStream()`.
+ * This name must be unique among all the currently active queries in the associated SQLContext.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def queryName(queryName: String): DataStreamWriter[T] = {
+ this.extraOptions += ("queryName" -> queryName)
+ this
+ }
+
+ /**
+ * :: Experimental ::
+ * Specifies the underlying output data source. Built-in options include "parquet", "json", etc.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def format(source: String): DataStreamWriter[T] = {
+ this.source = source
+ this
+ }
+
+ /**
+ * Partitions the output by the given columns on the file system. If specified, the output is
+ * laid out on the file system similar to Hive's partitioning scheme. As an example, when we
+ * partition a dataset by year and then month, the directory layout would look like:
+ *
+ * - year=2016/month=01/
+ * - year=2016/month=02/
+ *
+ * Partitioning is one of the most widely used techniques to optimize physical data layout.
+ * It provides a coarse-grained index for skipping unnecessary data reads when queries have
+ * predicates on the partitioned columns. In order for partitioning to work well, the number
+ * of distinct values in each column should typically be less than tens of thousands.
+ *
+ * This was initially applicable for Parquet but in 1.5+ covers JSON, text, ORC and avro as well.
+ *
+ * @since 1.4.0
+ */
+ @scala.annotation.varargs
+ def partitionBy(colNames: String*): DataStreamWriter[T] = {
+ this.partitioningColumns = Option(colNames)
+ this
+ }
+
+ /**
+ * :: Experimental ::
+ * Adds an output option for the underlying data source.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def option(key: String, value: String): DataStreamWriter[T] = {
+ this.extraOptions += (key -> value)
+ this
+ }
+
+ /**
+ * :: Experimental ::
+ * Adds an output option for the underlying data source.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def option(key: String, value: Boolean): DataStreamWriter[T] = option(key, value.toString)
+
+ /**
+ * :: Experimental ::
+ * Adds an output option for the underlying data source.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def option(key: String, value: Long): DataStreamWriter[T] = option(key, value.toString)
+
+ /**
+ * :: Experimental ::
+ * Adds an output option for the underlying data source.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def option(key: String, value: Double): DataStreamWriter[T] = option(key, value.toString)
+
+ /**
+ * :: Experimental ::
+ * (Scala-specific) Adds output options for the underlying data source.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def options(options: scala.collection.Map[String, String]): DataStreamWriter[T] = {
+ this.extraOptions ++= options
+ this
+ }
+
+ /**
+ * :: Experimental ::
+ * Adds output options for the underlying data source.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def options(options: java.util.Map[String, String]): DataStreamWriter[T] = {
+ this.options(options.asScala)
+ this
+ }
+
+
+ /**
+ * :: Experimental ::
+ * Saves the content of the streaming [[DataFrame]] in Parquet format at the specified path.
+ * This is equivalent to:
+ * {{{
+ * format("parquet").save(path)
+ * }}}
+ *
+ * You can set the following Parquet-specific option(s) for writing Parquet files:
+ * <li>`compression` (default is the value specified in `spark.sql.parquet.compression.codec`):
+ * compression codec to use when saving to file. This can be one of the known case-insensitive
+ * shorten names(none, `snappy`, `gzip`, and `lzo`). This will override
+ * `spark.sql.parquet.compression.codec`.</li>
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def parquet(path: String): ContinuousQuery = {
+ format("parquet").save(path)
+ }
+
+
+ /**
+ * :: Experimental ::
+ * Starts the execution of the streaming query, which will continually output results to the given
+ * path as new data arrives. The returned [[ContinuousQuery]] object can be used to interact with
+ * the stream.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def save(path: String): ContinuousQuery = {
--- End diff --
Here I did like `start()` since its much different than a single `save()` operation (i.e. its async and keeps going).
---
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 #13653: [SPARK-15933][SQL][STREAMING] Refactored DF reade...
Posted by tdas <gi...@git.apache.org>.
Github user tdas commented on a diff in the pull request:
https://github.com/apache/spark/pull/13653#discussion_r66888549
--- Diff: sql/core/src/test/scala/org/apache/spark/sql/test/DataFrameReaderWriterSuite.scala ---
@@ -0,0 +1,231 @@
+/*
+ * 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.test
+
+import org.apache.spark.sql._
+import org.apache.spark.sql.sources._
+import org.apache.spark.sql.types.{StringType, StructField, StructType}
+import org.apache.spark.util.Utils
+
+
+object LastOptions {
+
+ var parameters: Map[String, String] = null
+ var schema: Option[StructType] = null
+ var saveMode: SaveMode = null
+
+ def clear(): Unit = {
+ parameters = null
+ schema = null
+ saveMode = null
+ }
+}
+
+
+/** Dummy provider. */
+class DefaultSource
+ extends RelationProvider
+ with SchemaRelationProvider
+ with CreatableRelationProvider {
+
+ case class FakeRelation(sqlContext: SQLContext) extends BaseRelation {
+ override def schema: StructType = StructType(Seq(StructField("a", StringType)))
+ }
+
+ override def createRelation(
+ sqlContext: SQLContext,
+ parameters: Map[String, String],
+ schema: StructType
+ ): BaseRelation = {
+ LastOptions.parameters = parameters
+ LastOptions.schema = Some(schema)
+ FakeRelation(sqlContext)
+ }
+
+ override def createRelation(
+ sqlContext: SQLContext,
+ parameters: Map[String, String]
+ ): BaseRelation = {
+ LastOptions.parameters = parameters
+ LastOptions.schema = None
+ FakeRelation(sqlContext)
+ }
+
+ override def createRelation(
+ sqlContext: SQLContext,
+ mode: SaveMode,
+ parameters: Map[String, String],
+ data: DataFrame): BaseRelation = {
+ LastOptions.parameters = parameters
+ LastOptions.schema = None
+ LastOptions.saveMode = mode
+ FakeRelation(sqlContext)
+ }
+}
+
+
+class DataFrameReaderWriterSuite extends QueryTest with SharedSQLContext {
--- End diff --
These tests are very rudimentary set of tests to test the DataFrameReader code path. Note that in Spark 1.6, there were no tests at all. Also I believe that a lot of the DF functionality is tested through other test suites (e.g. partitioning columns is tested through PartitionedParquetSuite)
---
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 #13653: [SPARK-15933][SQL][STREAMING] Refactored DF reade...
Posted by tdas <gi...@git.apache.org>.
Github user tdas commented on a diff in the pull request:
https://github.com/apache/spark/pull/13653#discussion_r67029246
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/streaming/DataStreamWriter.scala ---
@@ -0,0 +1,401 @@
+/*
+ * 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.streaming
+
+import scala.collection.JavaConverters._
+
+import org.apache.spark.annotation.Experimental
+import org.apache.spark.sql.{AnalysisException, DataFrame, Dataset, ForeachWriter}
+import org.apache.spark.sql.execution.datasources.DataSource
+import org.apache.spark.sql.execution.streaming.{ForeachSink, MemoryPlan, MemorySink}
+
+/**
+ * :: Experimental ::
+ * Interface used to write a streaming [[Dataset]] to external storage systems (e.g. file systems,
+ * key-value stores, etc). Use [[Dataset.write]] to access this.
+ *
+ * @since 2.0.0
+ */
+@Experimental
+final class DataStreamWriter[T] private[sql](ds: Dataset[T]) {
+
+ private val df = ds.toDF()
+
+ /**
+ * :: Experimental ::
+ * Specifies how data of a streaming DataFrame/Dataset is written to a streaming sink.
+ * - `OutputMode.Append()`: only the new rows in the streaming DataFrame/Dataset will be
+ * written to the sink
+ * - `OutputMode.Complete()`: all the rows in the streaming DataFrame/Dataset will be written
+ * to the sink every time these is some updates
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def outputMode(outputMode: OutputMode): DataStreamWriter[T] = {
+ this.outputMode = outputMode
+ this
+ }
+
+
+ /**
+ * :: Experimental ::
+ * Specifies how data of a streaming DataFrame/Dataset is written to a streaming sink.
+ * - `append`: only the new rows in the streaming DataFrame/Dataset will be written to
+ * the sink
+ * - `complete`: all the rows in the streaming DataFrame/Dataset will be written to the sink
+ * every time these is some updates
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def outputMode(outputMode: String): DataStreamWriter[T] = {
+ this.outputMode = outputMode.toLowerCase match {
+ case "append" =>
+ OutputMode.Append
+ case "complete" =>
+ OutputMode.Complete
+ case _ =>
+ throw new IllegalArgumentException(s"Unknown output mode $outputMode. " +
+ "Accepted output modes are 'append' and 'complete'")
+ }
+ this
+ }
+
+ /**
+ * :: Experimental ::
+ * Set the trigger for the stream query. The default value is `ProcessingTime(0)` and it will run
+ * the query as fast as possible.
+ *
+ * Scala Example:
+ * {{{
+ * df.write.trigger(ProcessingTime("10 seconds"))
+ *
+ * import scala.concurrent.duration._
+ * df.write.trigger(ProcessingTime(10.seconds))
+ * }}}
+ *
+ * Java Example:
+ * {{{
+ * df.write.trigger(ProcessingTime.create("10 seconds"))
+ *
+ * import java.util.concurrent.TimeUnit
+ * df.write.trigger(ProcessingTime.create(10, TimeUnit.SECONDS))
+ * }}}
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def trigger(trigger: Trigger): DataStreamWriter[T] = {
+ this.trigger = trigger
+ this
+ }
+
+
+ /**
+ * :: Experimental ::
+ * Specifies the name of the [[ContinuousQuery]] that can be started with `startStream()`.
+ * This name must be unique among all the currently active queries in the associated SQLContext.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def queryName(queryName: String): DataStreamWriter[T] = {
+ this.extraOptions += ("queryName" -> queryName)
+ this
+ }
+
+ /**
+ * :: Experimental ::
+ * Specifies the underlying output data source. Built-in options include "parquet", "json", etc.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def format(source: String): DataStreamWriter[T] = {
+ this.source = source
+ this
+ }
+
+ /**
+ * Partitions the output by the given columns on the file system. If specified, the output is
+ * laid out on the file system similar to Hive's partitioning scheme. As an example, when we
+ * partition a dataset by year and then month, the directory layout would look like:
+ *
+ * - year=2016/month=01/
+ * - year=2016/month=02/
+ *
+ * Partitioning is one of the most widely used techniques to optimize physical data layout.
+ * It provides a coarse-grained index for skipping unnecessary data reads when queries have
+ * predicates on the partitioned columns. In order for partitioning to work well, the number
+ * of distinct values in each column should typically be less than tens of thousands.
+ *
+ * This was initially applicable for Parquet but in 1.5+ covers JSON, text, ORC and avro as well.
+ *
+ * @since 1.4.0
+ */
+ @scala.annotation.varargs
+ def partitionBy(colNames: String*): DataStreamWriter[T] = {
+ this.partitioningColumns = Option(colNames)
+ this
+ }
+
+ /**
+ * :: Experimental ::
+ * Adds an output option for the underlying data source.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def option(key: String, value: String): DataStreamWriter[T] = {
+ this.extraOptions += (key -> value)
+ this
+ }
+
+ /**
+ * :: Experimental ::
+ * Adds an output option for the underlying data source.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def option(key: String, value: Boolean): DataStreamWriter[T] = option(key, value.toString)
+
+ /**
+ * :: Experimental ::
+ * Adds an output option for the underlying data source.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def option(key: String, value: Long): DataStreamWriter[T] = option(key, value.toString)
+
+ /**
+ * :: Experimental ::
+ * Adds an output option for the underlying data source.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def option(key: String, value: Double): DataStreamWriter[T] = option(key, value.toString)
+
+ /**
+ * :: Experimental ::
+ * (Scala-specific) Adds output options for the underlying data source.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def options(options: scala.collection.Map[String, String]): DataStreamWriter[T] = {
+ this.extraOptions ++= options
+ this
+ }
+
+ /**
+ * :: Experimental ::
+ * Adds output options for the underlying data source.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def options(options: java.util.Map[String, String]): DataStreamWriter[T] = {
+ this.options(options.asScala)
+ this
+ }
+
+
+ /**
+ * :: Experimental ::
+ * Saves the content of the streaming [[DataFrame]] in Parquet format at the specified path.
+ * This is equivalent to:
+ * {{{
+ * format("parquet").save(path)
+ * }}}
+ *
+ * You can set the following Parquet-specific option(s) for writing Parquet files:
+ * <li>`compression` (default is the value specified in `spark.sql.parquet.compression.codec`):
+ * compression codec to use when saving to file. This can be one of the known case-insensitive
+ * shorten names(none, `snappy`, `gzip`, and `lzo`). This will override
+ * `spark.sql.parquet.compression.codec`.</li>
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def parquet(path: String): ContinuousQuery = {
+ format("parquet").save(path)
+ }
+
+
+ /**
+ * :: Experimental ::
+ * Starts the execution of the streaming query, which will continually output results to the given
+ * path as new data arrives. The returned [[ContinuousQuery]] object can be used to interact with
+ * the stream.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def save(path: String): ContinuousQuery = {
--- End diff --
Yes. I agree too. But then by that logic. Should `parquet()` automatically start the query?
---
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 issue #13653: [SPARK-15933][SQL][STREAMING] Refactored DF reader-write...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/13653
**[Test build #60519 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/60519/consoleFull)** for PR 13653 at commit [`29ca23b`](https://github.com/apache/spark/commit/29ca23b74c16440dbb1e440423615357ec8015bf).
* 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 #13653: [SPARK-15933][SQL][STREAMING] Refactored DF reade...
Posted by tdas <gi...@git.apache.org>.
Github user tdas commented on a diff in the pull request:
https://github.com/apache/spark/pull/13653#discussion_r67086347
--- Diff: python/pyspark/sql/readwriter.py ---
@@ -905,6 +764,503 @@ def jdbc(self, url, table, mode=None, properties=None):
self._jwrite.mode(mode).jdbc(url, table, jprop)
+class DataStreamReader(object):
--- End diff --
Done in #13673
---
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 #13653: [SPARK-15933][SQL][STREAMING] Refactored DF reade...
Posted by tdas <gi...@git.apache.org>.
Github user tdas commented on a diff in the pull request:
https://github.com/apache/spark/pull/13653#discussion_r67080464
--- Diff: python/pyspark/sql/readwriter.py ---
@@ -905,6 +764,503 @@ def jdbc(self, url, table, mode=None, properties=None):
self._jwrite.mode(mode).jdbc(url, table, jprop)
+class DataStreamReader(object):
+ """
+ Interface used to load a streaming :class:`DataFrame` from external storage systems
+ (e.g. file systems, key-value stores, etc). Use :func:`spark.readStream`
+ to access this.
+
+ .. note:: Experimental.
+
+ .. versionadded:: 2.0
+ """
+
+ def __init__(self, spark):
+ self._jreader = spark._ssql_ctx.readStream()
+ self._spark = spark
+
+ def _df(self, jdf):
+ from pyspark.sql.dataframe import DataFrame
+ return DataFrame(jdf, self._spark)
+
+ @since(2.0)
+ def format(self, source):
+ """Specifies the input data source format.
+
+ .. note:: Experimental.
+
+ :param source: string, name of the data source, e.g. 'json', 'parquet'.
+
+ """
+ self._jreader = self._jreader.format(source)
+ return self
+
+ @since(2.0)
+ def schema(self, schema):
+ """Specifies the input schema.
+
+ Some data sources (e.g. JSON) can infer the input schema automatically from data.
+ By specifying the schema here, the underlying data source can skip the schema
+ inference step, and thus speed up data loading.
+
+ .. note:: Experimental.
+
+ :param schema: a StructType object
+ """
+ if not isinstance(schema, StructType):
+ raise TypeError("schema should be StructType")
+ jschema = self._spark._ssql_ctx.parseDataType(schema.json())
+ self._jreader = self._jreader.schema(jschema)
+ return self
+
+ @since(2.0)
+ def option(self, key, value):
+ """Adds an input option for the underlying data source.
+
+ .. note:: Experimental.
+ """
+ self._jreader = self._jreader.option(key, to_str(value))
+ return self
+
+ @since(2.0)
+ def options(self, **options):
+ """Adds input options for the underlying data source.
+
+ .. note:: Experimental.
+ """
+ for k in options:
+ self._jreader = self._jreader.option(k, to_str(options[k]))
+ return self
+
+ @since(2.0)
+ def load(self, path=None, format=None, schema=None, **options):
+ """Loads a data stream from a data source and returns it as a :class`DataFrame`.
+
+ .. note:: Experimental.
+
+ :param path: optional string for file-system backed data sources.
+ :param format: optional string for format of the data source. Default to 'parquet'.
+ :param schema: optional :class:`StructType` for the input schema.
+ :param options: all other string options
+
+ """
+ if format is not None:
+ self.format(format)
+ if schema is not None:
+ self.schema(schema)
+ self.options(**options)
+ if path is not None:
+ if type(path) != str or len(path.strip()) == 0:
+ raise ValueError("If the path is provided for stream, it needs to be a " +
+ "non-empty string. List of paths are not supported.")
+ return self._df(self._jreader.load(path))
+ else:
+ return self._df(self._jreader.load())
+
+ @since(2.0)
+ def json(self, path, schema=None, primitivesAsString=None, prefersDecimal=None,
+ allowComments=None, allowUnquotedFieldNames=None, allowSingleQuotes=None,
+ allowNumericLeadingZero=None, allowBackslashEscapingAnyCharacter=None,
+ mode=None, columnNameOfCorruptRecord=None):
+ """
+ Loads a JSON file stream (one object per line) and returns a :class`DataFrame`.
+
+ If the ``schema`` parameter is not specified, this function goes
+ through the input once to determine the input schema.
+
+ .. note:: Experimental.
+
+ :param path: string represents path to the JSON dataset,
+ or RDD of Strings storing JSON objects.
+ :param schema: an optional :class:`StructType` for the input schema.
+ :param primitivesAsString: infers all primitive values as a string type. If None is set,
+ it uses the default value, ``false``.
+ :param prefersDecimal: infers all floating-point values as a decimal type. If the values
+ do not fit in decimal, then it infers them as doubles. If None is
+ set, it uses the default value, ``false``.
+ :param allowComments: ignores Java/C++ style comment in JSON records. If None is set,
+ it uses the default value, ``false``.
+ :param allowUnquotedFieldNames: allows unquoted JSON field names. If None is set,
+ it uses the default value, ``false``.
+ :param allowSingleQuotes: allows single quotes in addition to double quotes. If None is
+ set, it uses the default value, ``true``.
+ :param allowNumericLeadingZero: allows leading zeros in numbers (e.g. 00012). If None is
+ set, it uses the default value, ``false``.
+ :param allowBackslashEscapingAnyCharacter: allows accepting quoting of all character
+ using backslash quoting mechanism. If None is
+ set, it uses the default value, ``false``.
+ :param mode: allows a mode for dealing with corrupt records during parsing. If None is
+ set, it uses the default value, ``PERMISSIVE``.
+
+ * ``PERMISSIVE`` : sets other fields to ``null`` when it meets a corrupted \
+ record and puts the malformed string into a new field configured by \
+ ``columnNameOfCorruptRecord``. When a schema is set by user, it sets \
+ ``null`` for extra fields.
+ * ``DROPMALFORMED`` : ignores the whole corrupted records.
+ * ``FAILFAST`` : throws an exception when it meets corrupted records.
+
+ :param columnNameOfCorruptRecord: allows renaming the new field having malformed string
+ created by ``PERMISSIVE`` mode. This overrides
+ ``spark.sql.columnNameOfCorruptRecord``. If None is set,
+ it uses the value specified in
+ ``spark.sql.columnNameOfCorruptRecord``.
+
+ """
+ if schema is not None:
+ self.schema(schema)
+ if primitivesAsString is not None:
+ self.option("primitivesAsString", primitivesAsString)
+ if prefersDecimal is not None:
+ self.option("prefersDecimal", prefersDecimal)
+ if allowComments is not None:
+ self.option("allowComments", allowComments)
+ if allowUnquotedFieldNames is not None:
+ self.option("allowUnquotedFieldNames", allowUnquotedFieldNames)
+ if allowSingleQuotes is not None:
+ self.option("allowSingleQuotes", allowSingleQuotes)
+ if allowNumericLeadingZero is not None:
+ self.option("allowNumericLeadingZero", allowNumericLeadingZero)
+ if allowBackslashEscapingAnyCharacter is not None:
+ self.option("allowBackslashEscapingAnyCharacter", allowBackslashEscapingAnyCharacter)
+ if mode is not None:
+ self.option("mode", mode)
+ if columnNameOfCorruptRecord is not None:
+ self.option("columnNameOfCorruptRecord", columnNameOfCorruptRecord)
+ if isinstance(path, basestring):
+ path = [path]
+ return self._df(self._jreader.json(path))
+ else:
+ raise TypeError("path can be only a single string")
+
+ @since(2.0)
+ def parquet(self, path):
+ """Loads a Parquet file stream, returning the result as a :class:`DataFrame`.
+
+ You can set the following Parquet-specific option(s) for reading Parquet files:
+ * ``mergeSchema``: sets whether we should merge schemas collected from all \
+ Parquet part-files. This will override ``spark.sql.parquet.mergeSchema``. \
+ The default value is specified in ``spark.sql.parquet.mergeSchema``.
+
+ .. note:: Experimental.
+
+ """
+ if isinstance(path, basestring):
+ path = [path]
+ return self._df(self._jreader.parquet(self._spark._sc._jvm.PythonUtils.toSeq(path)))
+ else:
+ raise TypeError("path can be only a single string")
+
+ @ignore_unicode_prefix
+ @since(2.0)
+ def text(self, path):
+ """
+ Loads a text file stream and returns a :class:`DataFrame` whose schema starts with a
+ string column named "value", and followed by partitioned columns if there
+ are any.
+
+ Each line in the text file is a new row in the resulting DataFrame.
+
+ .. note:: Experimental.
+
+ :param paths: string, or list of strings, for input path(s).
+
+ """
+ if isinstance(path, basestring):
+ path = [path]
+ return self._df(self._jreader.text(self._spark._sc._jvm.PythonUtils.toSeq(path)))
+ else:
+ raise TypeError("path can be only a single string")
+
+ @since(2.0)
+ def csv(self, path, schema=None, sep=None, encoding=None, quote=None, escape=None,
--- End diff --
I would like to do that in that in a future PR. I would like to get this in as fast as possible.
---
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 #13653: [SPARK-15933][SQL][STREAMING] Refactored DF reade...
Posted by zsxwing <gi...@git.apache.org>.
Github user zsxwing commented on a diff in the pull request:
https://github.com/apache/spark/pull/13653#discussion_r67079701
--- Diff: python/pyspark/sql/readwriter.py ---
@@ -905,6 +764,503 @@ def jdbc(self, url, table, mode=None, properties=None):
self._jwrite.mode(mode).jdbc(url, table, jprop)
+class DataStreamReader(object):
+ """
+ Interface used to load a streaming :class:`DataFrame` from external storage systems
+ (e.g. file systems, key-value stores, etc). Use :func:`spark.readStream`
+ to access this.
+
+ .. note:: Experimental.
+
+ .. versionadded:: 2.0
+ """
+
+ def __init__(self, spark):
+ self._jreader = spark._ssql_ctx.readStream()
+ self._spark = spark
+
+ def _df(self, jdf):
+ from pyspark.sql.dataframe import DataFrame
+ return DataFrame(jdf, self._spark)
+
+ @since(2.0)
+ def format(self, source):
+ """Specifies the input data source format.
+
+ .. note:: Experimental.
+
+ :param source: string, name of the data source, e.g. 'json', 'parquet'.
+
+ """
+ self._jreader = self._jreader.format(source)
+ return self
+
+ @since(2.0)
+ def schema(self, schema):
+ """Specifies the input schema.
+
+ Some data sources (e.g. JSON) can infer the input schema automatically from data.
+ By specifying the schema here, the underlying data source can skip the schema
+ inference step, and thus speed up data loading.
+
+ .. note:: Experimental.
+
+ :param schema: a StructType object
+ """
+ if not isinstance(schema, StructType):
+ raise TypeError("schema should be StructType")
+ jschema = self._spark._ssql_ctx.parseDataType(schema.json())
+ self._jreader = self._jreader.schema(jschema)
+ return self
+
+ @since(2.0)
+ def option(self, key, value):
+ """Adds an input option for the underlying data source.
+
+ .. note:: Experimental.
+ """
+ self._jreader = self._jreader.option(key, to_str(value))
+ return self
+
+ @since(2.0)
+ def options(self, **options):
+ """Adds input options for the underlying data source.
+
+ .. note:: Experimental.
+ """
+ for k in options:
+ self._jreader = self._jreader.option(k, to_str(options[k]))
+ return self
+
+ @since(2.0)
+ def load(self, path=None, format=None, schema=None, **options):
+ """Loads a data stream from a data source and returns it as a :class`DataFrame`.
+
+ .. note:: Experimental.
+
+ :param path: optional string for file-system backed data sources.
+ :param format: optional string for format of the data source. Default to 'parquet'.
+ :param schema: optional :class:`StructType` for the input schema.
+ :param options: all other string options
+
+ """
+ if format is not None:
+ self.format(format)
+ if schema is not None:
+ self.schema(schema)
+ self.options(**options)
+ if path is not None:
+ if type(path) != str or len(path.strip()) == 0:
+ raise ValueError("If the path is provided for stream, it needs to be a " +
+ "non-empty string. List of paths are not supported.")
+ return self._df(self._jreader.load(path))
+ else:
+ return self._df(self._jreader.load())
+
+ @since(2.0)
+ def json(self, path, schema=None, primitivesAsString=None, prefersDecimal=None,
+ allowComments=None, allowUnquotedFieldNames=None, allowSingleQuotes=None,
+ allowNumericLeadingZero=None, allowBackslashEscapingAnyCharacter=None,
+ mode=None, columnNameOfCorruptRecord=None):
+ """
+ Loads a JSON file stream (one object per line) and returns a :class`DataFrame`.
+
+ If the ``schema`` parameter is not specified, this function goes
+ through the input once to determine the input schema.
+
+ .. note:: Experimental.
+
+ :param path: string represents path to the JSON dataset,
+ or RDD of Strings storing JSON objects.
+ :param schema: an optional :class:`StructType` for the input schema.
+ :param primitivesAsString: infers all primitive values as a string type. If None is set,
+ it uses the default value, ``false``.
+ :param prefersDecimal: infers all floating-point values as a decimal type. If the values
+ do not fit in decimal, then it infers them as doubles. If None is
+ set, it uses the default value, ``false``.
+ :param allowComments: ignores Java/C++ style comment in JSON records. If None is set,
+ it uses the default value, ``false``.
+ :param allowUnquotedFieldNames: allows unquoted JSON field names. If None is set,
+ it uses the default value, ``false``.
+ :param allowSingleQuotes: allows single quotes in addition to double quotes. If None is
+ set, it uses the default value, ``true``.
+ :param allowNumericLeadingZero: allows leading zeros in numbers (e.g. 00012). If None is
+ set, it uses the default value, ``false``.
+ :param allowBackslashEscapingAnyCharacter: allows accepting quoting of all character
+ using backslash quoting mechanism. If None is
+ set, it uses the default value, ``false``.
+ :param mode: allows a mode for dealing with corrupt records during parsing. If None is
+ set, it uses the default value, ``PERMISSIVE``.
+
+ * ``PERMISSIVE`` : sets other fields to ``null`` when it meets a corrupted \
+ record and puts the malformed string into a new field configured by \
+ ``columnNameOfCorruptRecord``. When a schema is set by user, it sets \
+ ``null`` for extra fields.
+ * ``DROPMALFORMED`` : ignores the whole corrupted records.
+ * ``FAILFAST`` : throws an exception when it meets corrupted records.
+
+ :param columnNameOfCorruptRecord: allows renaming the new field having malformed string
+ created by ``PERMISSIVE`` mode. This overrides
+ ``spark.sql.columnNameOfCorruptRecord``. If None is set,
+ it uses the value specified in
+ ``spark.sql.columnNameOfCorruptRecord``.
+
+ """
+ if schema is not None:
+ self.schema(schema)
+ if primitivesAsString is not None:
+ self.option("primitivesAsString", primitivesAsString)
+ if prefersDecimal is not None:
+ self.option("prefersDecimal", prefersDecimal)
+ if allowComments is not None:
+ self.option("allowComments", allowComments)
+ if allowUnquotedFieldNames is not None:
+ self.option("allowUnquotedFieldNames", allowUnquotedFieldNames)
+ if allowSingleQuotes is not None:
+ self.option("allowSingleQuotes", allowSingleQuotes)
+ if allowNumericLeadingZero is not None:
+ self.option("allowNumericLeadingZero", allowNumericLeadingZero)
+ if allowBackslashEscapingAnyCharacter is not None:
+ self.option("allowBackslashEscapingAnyCharacter", allowBackslashEscapingAnyCharacter)
+ if mode is not None:
+ self.option("mode", mode)
+ if columnNameOfCorruptRecord is not None:
+ self.option("columnNameOfCorruptRecord", columnNameOfCorruptRecord)
+ if isinstance(path, basestring):
+ path = [path]
+ return self._df(self._jreader.json(path))
+ else:
+ raise TypeError("path can be only a single string")
+
+ @since(2.0)
+ def parquet(self, path):
+ """Loads a Parquet file stream, returning the result as a :class:`DataFrame`.
+
+ You can set the following Parquet-specific option(s) for reading Parquet files:
+ * ``mergeSchema``: sets whether we should merge schemas collected from all \
+ Parquet part-files. This will override ``spark.sql.parquet.mergeSchema``. \
+ The default value is specified in ``spark.sql.parquet.mergeSchema``.
+
+ .. note:: Experimental.
+
+ """
+ if isinstance(path, basestring):
+ path = [path]
+ return self._df(self._jreader.parquet(self._spark._sc._jvm.PythonUtils.toSeq(path)))
+ else:
+ raise TypeError("path can be only a single string")
+
+ @ignore_unicode_prefix
+ @since(2.0)
+ def text(self, path):
+ """
+ Loads a text file stream and returns a :class:`DataFrame` whose schema starts with a
+ string column named "value", and followed by partitioned columns if there
+ are any.
+
+ Each line in the text file is a new row in the resulting DataFrame.
+
+ .. note:: Experimental.
+
+ :param paths: string, or list of strings, for input path(s).
+
+ """
+ if isinstance(path, basestring):
+ path = [path]
+ return self._df(self._jreader.text(self._spark._sc._jvm.PythonUtils.toSeq(path)))
+ else:
+ raise TypeError("path can be only a single string")
+
+ @since(2.0)
+ def csv(self, path, schema=None, sep=None, encoding=None, quote=None, escape=None,
--- End diff --
I'm wondering if we can create util methods for `csv` and `json` to make DataFrameReader and DataStreamReader share most of codes here. Otherwise, it's hard to make them consistent in future.
---
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 issue #13653: [SPARK-15933][SQL][STREAMING] Refactored DF reader-write...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/13653
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/60519/
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 issue #13653: [SPARK-15933][SQL][STREAMING] Refactored DF reader-write...
Posted by marmbrus <gi...@git.apache.org>.
Github user marmbrus commented on the issue:
https://github.com/apache/spark/pull/13653
Overall looks pretty good. Feel free to merge after addressing comments / passing tests to avoid more conflicts.
---
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 issue #13653: [SPARK-15933][SQL][STREAMING] Refactored DF reader-write...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/13653
**[Test build #60519 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/60519/consoleFull)** for PR 13653 at commit [`29ca23b`](https://github.com/apache/spark/commit/29ca23b74c16440dbb1e440423615357ec8015bf).
---
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 #13653: [SPARK-15933][SQL][STREAMING] Refactored DF reade...
Posted by marmbrus <gi...@git.apache.org>.
Github user marmbrus commented on a diff in the pull request:
https://github.com/apache/spark/pull/13653#discussion_r67029686
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/streaming/DataStreamWriter.scala ---
@@ -0,0 +1,401 @@
+/*
+ * 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.streaming
+
+import scala.collection.JavaConverters._
+
+import org.apache.spark.annotation.Experimental
+import org.apache.spark.sql.{AnalysisException, DataFrame, Dataset, ForeachWriter}
+import org.apache.spark.sql.execution.datasources.DataSource
+import org.apache.spark.sql.execution.streaming.{ForeachSink, MemoryPlan, MemorySink}
+
+/**
+ * :: Experimental ::
+ * Interface used to write a streaming [[Dataset]] to external storage systems (e.g. file systems,
+ * key-value stores, etc). Use [[Dataset.write]] to access this.
+ *
+ * @since 2.0.0
+ */
+@Experimental
+final class DataStreamWriter[T] private[sql](ds: Dataset[T]) {
+
+ private val df = ds.toDF()
+
+ /**
+ * :: Experimental ::
+ * Specifies how data of a streaming DataFrame/Dataset is written to a streaming sink.
+ * - `OutputMode.Append()`: only the new rows in the streaming DataFrame/Dataset will be
+ * written to the sink
+ * - `OutputMode.Complete()`: all the rows in the streaming DataFrame/Dataset will be written
+ * to the sink every time these is some updates
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def outputMode(outputMode: OutputMode): DataStreamWriter[T] = {
+ this.outputMode = outputMode
+ this
+ }
+
+
+ /**
+ * :: Experimental ::
+ * Specifies how data of a streaming DataFrame/Dataset is written to a streaming sink.
+ * - `append`: only the new rows in the streaming DataFrame/Dataset will be written to
+ * the sink
+ * - `complete`: all the rows in the streaming DataFrame/Dataset will be written to the sink
+ * every time these is some updates
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def outputMode(outputMode: String): DataStreamWriter[T] = {
+ this.outputMode = outputMode.toLowerCase match {
+ case "append" =>
+ OutputMode.Append
+ case "complete" =>
+ OutputMode.Complete
+ case _ =>
+ throw new IllegalArgumentException(s"Unknown output mode $outputMode. " +
+ "Accepted output modes are 'append' and 'complete'")
+ }
+ this
+ }
+
+ /**
+ * :: Experimental ::
+ * Set the trigger for the stream query. The default value is `ProcessingTime(0)` and it will run
+ * the query as fast as possible.
+ *
+ * Scala Example:
+ * {{{
+ * df.write.trigger(ProcessingTime("10 seconds"))
+ *
+ * import scala.concurrent.duration._
+ * df.write.trigger(ProcessingTime(10.seconds))
+ * }}}
+ *
+ * Java Example:
+ * {{{
+ * df.write.trigger(ProcessingTime.create("10 seconds"))
+ *
+ * import java.util.concurrent.TimeUnit
+ * df.write.trigger(ProcessingTime.create(10, TimeUnit.SECONDS))
+ * }}}
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def trigger(trigger: Trigger): DataStreamWriter[T] = {
+ this.trigger = trigger
+ this
+ }
+
+
+ /**
+ * :: Experimental ::
+ * Specifies the name of the [[ContinuousQuery]] that can be started with `startStream()`.
+ * This name must be unique among all the currently active queries in the associated SQLContext.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def queryName(queryName: String): DataStreamWriter[T] = {
+ this.extraOptions += ("queryName" -> queryName)
+ this
+ }
+
+ /**
+ * :: Experimental ::
+ * Specifies the underlying output data source. Built-in options include "parquet", "json", etc.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def format(source: String): DataStreamWriter[T] = {
+ this.source = source
+ this
+ }
+
+ /**
+ * Partitions the output by the given columns on the file system. If specified, the output is
+ * laid out on the file system similar to Hive's partitioning scheme. As an example, when we
+ * partition a dataset by year and then month, the directory layout would look like:
+ *
+ * - year=2016/month=01/
+ * - year=2016/month=02/
+ *
+ * Partitioning is one of the most widely used techniques to optimize physical data layout.
+ * It provides a coarse-grained index for skipping unnecessary data reads when queries have
+ * predicates on the partitioned columns. In order for partitioning to work well, the number
+ * of distinct values in each column should typically be less than tens of thousands.
+ *
+ * This was initially applicable for Parquet but in 1.5+ covers JSON, text, ORC and avro as well.
+ *
+ * @since 1.4.0
+ */
+ @scala.annotation.varargs
+ def partitionBy(colNames: String*): DataStreamWriter[T] = {
+ this.partitioningColumns = Option(colNames)
+ this
+ }
+
+ /**
+ * :: Experimental ::
+ * Adds an output option for the underlying data source.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def option(key: String, value: String): DataStreamWriter[T] = {
+ this.extraOptions += (key -> value)
+ this
+ }
+
+ /**
+ * :: Experimental ::
+ * Adds an output option for the underlying data source.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def option(key: String, value: Boolean): DataStreamWriter[T] = option(key, value.toString)
+
+ /**
+ * :: Experimental ::
+ * Adds an output option for the underlying data source.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def option(key: String, value: Long): DataStreamWriter[T] = option(key, value.toString)
+
+ /**
+ * :: Experimental ::
+ * Adds an output option for the underlying data source.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def option(key: String, value: Double): DataStreamWriter[T] = option(key, value.toString)
+
+ /**
+ * :: Experimental ::
+ * (Scala-specific) Adds output options for the underlying data source.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def options(options: scala.collection.Map[String, String]): DataStreamWriter[T] = {
+ this.extraOptions ++= options
+ this
+ }
+
+ /**
+ * :: Experimental ::
+ * Adds output options for the underlying data source.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def options(options: java.util.Map[String, String]): DataStreamWriter[T] = {
+ this.options(options.asScala)
+ this
+ }
+
+
+ /**
+ * :: Experimental ::
+ * Saves the content of the streaming [[DataFrame]] in Parquet format at the specified path.
+ * This is equivalent to:
+ * {{{
+ * format("parquet").save(path)
+ * }}}
+ *
+ * You can set the following Parquet-specific option(s) for writing Parquet files:
+ * <li>`compression` (default is the value specified in `spark.sql.parquet.compression.codec`):
+ * compression codec to use when saving to file. This can be one of the known case-insensitive
+ * shorten names(none, `snappy`, `gzip`, and `lzo`). This will override
+ * `spark.sql.parquet.compression.codec`.</li>
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def parquet(path: String): ContinuousQuery = {
+ format("parquet").save(path)
+ }
+
+
+ /**
+ * :: Experimental ::
+ * Starts the execution of the streaming query, which will continually output results to the given
+ * path as new data arrives. The returned [[ContinuousQuery]] object can be used to interact with
+ * the stream.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def save(path: String): ContinuousQuery = {
--- End diff --
`writeStream.parquet()` doesn't seem incongruous with its behavior to me.
---
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 issue #13653: [SPARK-15933][SQL][STREAMING] Refactored DF reader-write...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/13653
**[Test build #60516 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/60516/consoleFull)** for PR 13653 at commit [`536c25e`](https://github.com/apache/spark/commit/536c25e00b154a467279ab0055ae98002bd70847).
* This patch **fails Python style 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 #13653: [SPARK-15933][SQL][STREAMING] Refactored DF reade...
Posted by tdas <gi...@git.apache.org>.
Github user tdas commented on a diff in the pull request:
https://github.com/apache/spark/pull/13653#discussion_r67042657
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/streaming/DataStreamWriter.scala ---
@@ -0,0 +1,401 @@
+/*
+ * 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.streaming
+
+import scala.collection.JavaConverters._
+
+import org.apache.spark.annotation.Experimental
+import org.apache.spark.sql.{AnalysisException, DataFrame, Dataset, ForeachWriter}
+import org.apache.spark.sql.execution.datasources.DataSource
+import org.apache.spark.sql.execution.streaming.{ForeachSink, MemoryPlan, MemorySink}
+
+/**
+ * :: Experimental ::
+ * Interface used to write a streaming [[Dataset]] to external storage systems (e.g. file systems,
+ * key-value stores, etc). Use [[Dataset.write]] to access this.
+ *
+ * @since 2.0.0
+ */
+@Experimental
+final class DataStreamWriter[T] private[sql](ds: Dataset[T]) {
+
+ private val df = ds.toDF()
+
+ /**
+ * :: Experimental ::
+ * Specifies how data of a streaming DataFrame/Dataset is written to a streaming sink.
+ * - `OutputMode.Append()`: only the new rows in the streaming DataFrame/Dataset will be
+ * written to the sink
+ * - `OutputMode.Complete()`: all the rows in the streaming DataFrame/Dataset will be written
+ * to the sink every time these is some updates
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def outputMode(outputMode: OutputMode): DataStreamWriter[T] = {
+ this.outputMode = outputMode
+ this
+ }
+
+
+ /**
+ * :: Experimental ::
+ * Specifies how data of a streaming DataFrame/Dataset is written to a streaming sink.
+ * - `append`: only the new rows in the streaming DataFrame/Dataset will be written to
+ * the sink
+ * - `complete`: all the rows in the streaming DataFrame/Dataset will be written to the sink
+ * every time these is some updates
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def outputMode(outputMode: String): DataStreamWriter[T] = {
+ this.outputMode = outputMode.toLowerCase match {
+ case "append" =>
+ OutputMode.Append
+ case "complete" =>
+ OutputMode.Complete
+ case _ =>
+ throw new IllegalArgumentException(s"Unknown output mode $outputMode. " +
+ "Accepted output modes are 'append' and 'complete'")
+ }
+ this
+ }
+
+ /**
+ * :: Experimental ::
+ * Set the trigger for the stream query. The default value is `ProcessingTime(0)` and it will run
+ * the query as fast as possible.
+ *
+ * Scala Example:
+ * {{{
+ * df.write.trigger(ProcessingTime("10 seconds"))
+ *
+ * import scala.concurrent.duration._
+ * df.write.trigger(ProcessingTime(10.seconds))
+ * }}}
+ *
+ * Java Example:
+ * {{{
+ * df.write.trigger(ProcessingTime.create("10 seconds"))
+ *
+ * import java.util.concurrent.TimeUnit
+ * df.write.trigger(ProcessingTime.create(10, TimeUnit.SECONDS))
+ * }}}
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def trigger(trigger: Trigger): DataStreamWriter[T] = {
+ this.trigger = trigger
+ this
+ }
+
+
+ /**
+ * :: Experimental ::
+ * Specifies the name of the [[ContinuousQuery]] that can be started with `startStream()`.
+ * This name must be unique among all the currently active queries in the associated SQLContext.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def queryName(queryName: String): DataStreamWriter[T] = {
+ this.extraOptions += ("queryName" -> queryName)
+ this
+ }
+
+ /**
+ * :: Experimental ::
+ * Specifies the underlying output data source. Built-in options include "parquet", "json", etc.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def format(source: String): DataStreamWriter[T] = {
+ this.source = source
+ this
+ }
+
+ /**
+ * Partitions the output by the given columns on the file system. If specified, the output is
+ * laid out on the file system similar to Hive's partitioning scheme. As an example, when we
+ * partition a dataset by year and then month, the directory layout would look like:
+ *
+ * - year=2016/month=01/
+ * - year=2016/month=02/
+ *
+ * Partitioning is one of the most widely used techniques to optimize physical data layout.
+ * It provides a coarse-grained index for skipping unnecessary data reads when queries have
+ * predicates on the partitioned columns. In order for partitioning to work well, the number
+ * of distinct values in each column should typically be less than tens of thousands.
+ *
+ * This was initially applicable for Parquet but in 1.5+ covers JSON, text, ORC and avro as well.
+ *
+ * @since 1.4.0
+ */
+ @scala.annotation.varargs
+ def partitionBy(colNames: String*): DataStreamWriter[T] = {
+ this.partitioningColumns = Option(colNames)
+ this
+ }
+
+ /**
+ * :: Experimental ::
+ * Adds an output option for the underlying data source.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def option(key: String, value: String): DataStreamWriter[T] = {
+ this.extraOptions += (key -> value)
+ this
+ }
+
+ /**
+ * :: Experimental ::
+ * Adds an output option for the underlying data source.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def option(key: String, value: Boolean): DataStreamWriter[T] = option(key, value.toString)
+
+ /**
+ * :: Experimental ::
+ * Adds an output option for the underlying data source.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def option(key: String, value: Long): DataStreamWriter[T] = option(key, value.toString)
+
+ /**
+ * :: Experimental ::
+ * Adds an output option for the underlying data source.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def option(key: String, value: Double): DataStreamWriter[T] = option(key, value.toString)
+
+ /**
+ * :: Experimental ::
+ * (Scala-specific) Adds output options for the underlying data source.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def options(options: scala.collection.Map[String, String]): DataStreamWriter[T] = {
+ this.extraOptions ++= options
+ this
+ }
+
+ /**
+ * :: Experimental ::
+ * Adds output options for the underlying data source.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def options(options: java.util.Map[String, String]): DataStreamWriter[T] = {
+ this.options(options.asScala)
+ this
+ }
+
+
+ /**
+ * :: Experimental ::
+ * Saves the content of the streaming [[DataFrame]] in Parquet format at the specified path.
+ * This is equivalent to:
+ * {{{
+ * format("parquet").save(path)
+ * }}}
+ *
+ * You can set the following Parquet-specific option(s) for writing Parquet files:
+ * <li>`compression` (default is the value specified in `spark.sql.parquet.compression.codec`):
+ * compression codec to use when saving to file. This can be one of the known case-insensitive
+ * shorten names(none, `snappy`, `gzip`, and `lzo`). This will override
+ * `spark.sql.parquet.compression.codec`.</li>
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def parquet(path: String): ContinuousQuery = {
+ format("parquet").save(path)
+ }
+
+
+ /**
+ * :: Experimental ::
+ * Starts the execution of the streaming query, which will continually output results to the given
+ * path as new data arrives. The returned [[ContinuousQuery]] object can be used to interact with
+ * the stream.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def save(path: String): ContinuousQuery = {
--- End diff --
I think I will remove parquet for now.
---
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 issue #13653: [SPARK-15933][SQL][STREAMING] Refactored DF reader-write...
Posted by tdas <gi...@git.apache.org>.
Github user tdas commented on the issue:
https://github.com/apache/spark/pull/13653
@marmbrus Overall, I have changed the following.
- Renamed writeStream.save() to writeStream.start()
- Refactored writeStream.foreach() to not start the query. instead writeStream.foreach().start() starts the query
- Removed writeStream.parquet(path) for now. Its good to have a single method `start` actually start the background query. experience says that its much easier read code and debug.
---
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 #13653: [SPARK-15933][SQL][STREAMING] Refactored DF reade...
Posted by tdas <gi...@git.apache.org>.
Github user tdas commented on a diff in the pull request:
https://github.com/apache/spark/pull/13653#discussion_r66888185
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/streaming/DataStreamWriter.scala ---
@@ -0,0 +1,401 @@
+/*
+ * 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.streaming
+
+import scala.collection.JavaConverters._
+
+import org.apache.spark.annotation.Experimental
+import org.apache.spark.sql.{AnalysisException, DataFrame, Dataset, ForeachWriter}
+import org.apache.spark.sql.execution.datasources.DataSource
+import org.apache.spark.sql.execution.streaming.{ForeachSink, MemoryPlan, MemorySink}
+
+/**
+ * :: Experimental ::
+ * Interface used to write a streaming [[Dataset]] to external storage systems (e.g. file systems,
+ * key-value stores, etc). Use [[Dataset.write]] to access this.
+ *
+ * @since 2.0.0
+ */
+@Experimental
+final class DataStreamWriter[T] private[sql](ds: Dataset[T]) {
+
+ private val df = ds.toDF()
+
+ /**
+ * :: Experimental ::
+ * Specifies how data of a streaming DataFrame/Dataset is written to a streaming sink.
+ * - `OutputMode.Append()`: only the new rows in the streaming DataFrame/Dataset will be
+ * written to the sink
+ * - `OutputMode.Complete()`: all the rows in the streaming DataFrame/Dataset will be written
+ * to the sink every time these is some updates
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def outputMode(outputMode: OutputMode): DataStreamWriter[T] = {
+ this.outputMode = outputMode
+ this
+ }
+
+
+ /**
+ * :: Experimental ::
+ * Specifies how data of a streaming DataFrame/Dataset is written to a streaming sink.
+ * - `append`: only the new rows in the streaming DataFrame/Dataset will be written to
+ * the sink
+ * - `complete`: all the rows in the streaming DataFrame/Dataset will be written to the sink
+ * every time these is some updates
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def outputMode(outputMode: String): DataStreamWriter[T] = {
+ this.outputMode = outputMode.toLowerCase match {
+ case "append" =>
+ OutputMode.Append
+ case "complete" =>
+ OutputMode.Complete
+ case _ =>
+ throw new IllegalArgumentException(s"Unknown output mode $outputMode. " +
+ "Accepted output modes are 'append' and 'complete'")
+ }
+ this
+ }
+
+ /**
+ * :: Experimental ::
+ * Set the trigger for the stream query. The default value is `ProcessingTime(0)` and it will run
+ * the query as fast as possible.
+ *
+ * Scala Example:
+ * {{{
+ * df.write.trigger(ProcessingTime("10 seconds"))
+ *
+ * import scala.concurrent.duration._
+ * df.write.trigger(ProcessingTime(10.seconds))
+ * }}}
+ *
+ * Java Example:
+ * {{{
+ * df.write.trigger(ProcessingTime.create("10 seconds"))
+ *
+ * import java.util.concurrent.TimeUnit
+ * df.write.trigger(ProcessingTime.create(10, TimeUnit.SECONDS))
+ * }}}
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def trigger(trigger: Trigger): DataStreamWriter[T] = {
+ this.trigger = trigger
+ this
+ }
+
+
+ /**
+ * :: Experimental ::
+ * Specifies the name of the [[ContinuousQuery]] that can be started with `startStream()`.
+ * This name must be unique among all the currently active queries in the associated SQLContext.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def queryName(queryName: String): DataStreamWriter[T] = {
+ this.extraOptions += ("queryName" -> queryName)
+ this
+ }
+
+ /**
+ * :: Experimental ::
+ * Specifies the underlying output data source. Built-in options include "parquet", "json", etc.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def format(source: String): DataStreamWriter[T] = {
+ this.source = source
+ this
+ }
+
+ /**
+ * Partitions the output by the given columns on the file system. If specified, the output is
+ * laid out on the file system similar to Hive's partitioning scheme. As an example, when we
+ * partition a dataset by year and then month, the directory layout would look like:
+ *
+ * - year=2016/month=01/
+ * - year=2016/month=02/
+ *
+ * Partitioning is one of the most widely used techniques to optimize physical data layout.
+ * It provides a coarse-grained index for skipping unnecessary data reads when queries have
+ * predicates on the partitioned columns. In order for partitioning to work well, the number
+ * of distinct values in each column should typically be less than tens of thousands.
+ *
+ * This was initially applicable for Parquet but in 1.5+ covers JSON, text, ORC and avro as well.
+ *
+ * @since 1.4.0
+ */
+ @scala.annotation.varargs
+ def partitionBy(colNames: String*): DataStreamWriter[T] = {
+ this.partitioningColumns = Option(colNames)
+ this
+ }
+
+ /**
+ * :: Experimental ::
+ * Adds an output option for the underlying data source.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def option(key: String, value: String): DataStreamWriter[T] = {
+ this.extraOptions += (key -> value)
+ this
+ }
+
+ /**
+ * :: Experimental ::
+ * Adds an output option for the underlying data source.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def option(key: String, value: Boolean): DataStreamWriter[T] = option(key, value.toString)
+
+ /**
+ * :: Experimental ::
+ * Adds an output option for the underlying data source.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def option(key: String, value: Long): DataStreamWriter[T] = option(key, value.toString)
+
+ /**
+ * :: Experimental ::
+ * Adds an output option for the underlying data source.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def option(key: String, value: Double): DataStreamWriter[T] = option(key, value.toString)
+
+ /**
+ * :: Experimental ::
+ * (Scala-specific) Adds output options for the underlying data source.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def options(options: scala.collection.Map[String, String]): DataStreamWriter[T] = {
+ this.extraOptions ++= options
+ this
+ }
+
+ /**
+ * :: Experimental ::
+ * Adds output options for the underlying data source.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def options(options: java.util.Map[String, String]): DataStreamWriter[T] = {
+ this.options(options.asScala)
+ this
+ }
+
+
+ /**
+ * :: Experimental ::
+ * Saves the content of the streaming [[DataFrame]] in Parquet format at the specified path.
+ * This is equivalent to:
+ * {{{
+ * format("parquet").save(path)
+ * }}}
+ *
+ * You can set the following Parquet-specific option(s) for writing Parquet files:
+ * <li>`compression` (default is the value specified in `spark.sql.parquet.compression.codec`):
+ * compression codec to use when saving to file. This can be one of the known case-insensitive
+ * shorten names(none, `snappy`, `gzip`, and `lzo`). This will override
+ * `spark.sql.parquet.compression.codec`.</li>
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def parquet(path: String): ContinuousQuery = {
+ format("parquet").save(path)
+ }
+
+
+ /**
+ * :: Experimental ::
+ * Starts the execution of the streaming query, which will continually output results to the given
+ * path as new data arrives. The returned [[ContinuousQuery]] object can be used to interact with
+ * the stream.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def save(path: String): ContinuousQuery = {
--- End diff --
@marmbrus @rxin @mateiz Should this be `save()`? The general case will be
```
val query = sdf
.writeStream
.format("myFormat")
.save("stringIdentifier")
```
---
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 issue #13653: [SPARK-15933][SQL][STREAMING] Refactored DF reader-write...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/13653
**[Test build #60460 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/60460/consoleFull)** for PR 13653 at commit [`e118631`](https://github.com/apache/spark/commit/e118631fd5bbea962932a812addae21c89f6f4f2).
---
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 issue #13653: [SPARK-15933][SQL][STREAMING] Refactored DF reader-write...
Posted by tdas <gi...@git.apache.org>.
Github user tdas commented on the issue:
https://github.com/apache/spark/pull/13653
I am merging this for now. Please keep reviewing this and I will update them in #13673
---
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 #13653: [SPARK-15933][SQL][STREAMING] Refactored DF reade...
Posted by tdas <gi...@git.apache.org>.
Github user tdas commented on a diff in the pull request:
https://github.com/apache/spark/pull/13653#discussion_r66888051
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/streaming/DataStreamWriter.scala ---
@@ -0,0 +1,401 @@
+/*
+ * 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.streaming
+
+import scala.collection.JavaConverters._
+
+import org.apache.spark.annotation.Experimental
+import org.apache.spark.sql.{AnalysisException, DataFrame, Dataset, ForeachWriter}
+import org.apache.spark.sql.execution.datasources.DataSource
+import org.apache.spark.sql.execution.streaming.{ForeachSink, MemoryPlan, MemorySink}
+
+/**
+ * :: Experimental ::
+ * Interface used to write a streaming [[Dataset]] to external storage systems (e.g. file systems,
+ * key-value stores, etc). Use [[Dataset.write]] to access this.
+ *
+ * @since 2.0.0
+ */
+@Experimental
+final class DataStreamWriter[T] private[sql](ds: Dataset[T]) {
+
+ private val df = ds.toDF()
+
+ /**
+ * :: Experimental ::
+ * Specifies how data of a streaming DataFrame/Dataset is written to a streaming sink.
+ * - `OutputMode.Append()`: only the new rows in the streaming DataFrame/Dataset will be
+ * written to the sink
+ * - `OutputMode.Complete()`: all the rows in the streaming DataFrame/Dataset will be written
+ * to the sink every time these is some updates
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def outputMode(outputMode: OutputMode): DataStreamWriter[T] = {
+ this.outputMode = outputMode
+ this
+ }
+
+
+ /**
+ * :: Experimental ::
+ * Specifies how data of a streaming DataFrame/Dataset is written to a streaming sink.
+ * - `append`: only the new rows in the streaming DataFrame/Dataset will be written to
+ * the sink
+ * - `complete`: all the rows in the streaming DataFrame/Dataset will be written to the sink
+ * every time these is some updates
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def outputMode(outputMode: String): DataStreamWriter[T] = {
+ this.outputMode = outputMode.toLowerCase match {
+ case "append" =>
+ OutputMode.Append
+ case "complete" =>
+ OutputMode.Complete
+ case _ =>
+ throw new IllegalArgumentException(s"Unknown output mode $outputMode. " +
+ "Accepted output modes are 'append' and 'complete'")
+ }
+ this
+ }
+
+ /**
+ * :: Experimental ::
+ * Set the trigger for the stream query. The default value is `ProcessingTime(0)` and it will run
+ * the query as fast as possible.
+ *
+ * Scala Example:
+ * {{{
+ * df.write.trigger(ProcessingTime("10 seconds"))
+ *
+ * import scala.concurrent.duration._
+ * df.write.trigger(ProcessingTime(10.seconds))
+ * }}}
+ *
+ * Java Example:
+ * {{{
+ * df.write.trigger(ProcessingTime.create("10 seconds"))
+ *
+ * import java.util.concurrent.TimeUnit
+ * df.write.trigger(ProcessingTime.create(10, TimeUnit.SECONDS))
+ * }}}
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def trigger(trigger: Trigger): DataStreamWriter[T] = {
+ this.trigger = trigger
+ this
+ }
+
+
+ /**
+ * :: Experimental ::
+ * Specifies the name of the [[ContinuousQuery]] that can be started with `startStream()`.
+ * This name must be unique among all the currently active queries in the associated SQLContext.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def queryName(queryName: String): DataStreamWriter[T] = {
+ this.extraOptions += ("queryName" -> queryName)
+ this
+ }
+
+ /**
+ * :: Experimental ::
+ * Specifies the underlying output data source. Built-in options include "parquet", "json", etc.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def format(source: String): DataStreamWriter[T] = {
+ this.source = source
+ this
+ }
+
+ /**
+ * Partitions the output by the given columns on the file system. If specified, the output is
+ * laid out on the file system similar to Hive's partitioning scheme. As an example, when we
+ * partition a dataset by year and then month, the directory layout would look like:
+ *
+ * - year=2016/month=01/
+ * - year=2016/month=02/
+ *
+ * Partitioning is one of the most widely used techniques to optimize physical data layout.
+ * It provides a coarse-grained index for skipping unnecessary data reads when queries have
+ * predicates on the partitioned columns. In order for partitioning to work well, the number
+ * of distinct values in each column should typically be less than tens of thousands.
+ *
+ * This was initially applicable for Parquet but in 1.5+ covers JSON, text, ORC and avro as well.
+ *
+ * @since 1.4.0
+ */
+ @scala.annotation.varargs
+ def partitionBy(colNames: String*): DataStreamWriter[T] = {
+ this.partitioningColumns = Option(colNames)
+ this
+ }
+
+ /**
+ * :: Experimental ::
+ * Adds an output option for the underlying data source.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def option(key: String, value: String): DataStreamWriter[T] = {
+ this.extraOptions += (key -> value)
+ this
+ }
+
+ /**
+ * :: Experimental ::
+ * Adds an output option for the underlying data source.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def option(key: String, value: Boolean): DataStreamWriter[T] = option(key, value.toString)
+
+ /**
+ * :: Experimental ::
+ * Adds an output option for the underlying data source.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def option(key: String, value: Long): DataStreamWriter[T] = option(key, value.toString)
+
+ /**
+ * :: Experimental ::
+ * Adds an output option for the underlying data source.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def option(key: String, value: Double): DataStreamWriter[T] = option(key, value.toString)
+
+ /**
+ * :: Experimental ::
+ * (Scala-specific) Adds output options for the underlying data source.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def options(options: scala.collection.Map[String, String]): DataStreamWriter[T] = {
+ this.extraOptions ++= options
+ this
+ }
+
+ /**
+ * :: Experimental ::
+ * Adds output options for the underlying data source.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def options(options: java.util.Map[String, String]): DataStreamWriter[T] = {
+ this.options(options.asScala)
+ this
+ }
+
+
+ /**
+ * :: Experimental ::
+ * Saves the content of the streaming [[DataFrame]] in Parquet format at the specified path.
+ * This is equivalent to:
+ * {{{
+ * format("parquet").save(path)
--- End diff --
update `save` if this is changed.
---
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 #13653: [SPARK-15933][SQL][STREAMING] Refactored DF reade...
Posted by tdas <gi...@git.apache.org>.
Github user tdas commented on a diff in the pull request:
https://github.com/apache/spark/pull/13653#discussion_r66888223
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/streaming/DataStreamWriter.scala ---
@@ -0,0 +1,401 @@
+/*
+ * 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.streaming
+
+import scala.collection.JavaConverters._
+
+import org.apache.spark.annotation.Experimental
+import org.apache.spark.sql.{AnalysisException, DataFrame, Dataset, ForeachWriter}
+import org.apache.spark.sql.execution.datasources.DataSource
+import org.apache.spark.sql.execution.streaming.{ForeachSink, MemoryPlan, MemorySink}
+
+/**
+ * :: Experimental ::
+ * Interface used to write a streaming [[Dataset]] to external storage systems (e.g. file systems,
+ * key-value stores, etc). Use [[Dataset.write]] to access this.
+ *
+ * @since 2.0.0
+ */
+@Experimental
+final class DataStreamWriter[T] private[sql](ds: Dataset[T]) {
+
+ private val df = ds.toDF()
+
+ /**
+ * :: Experimental ::
+ * Specifies how data of a streaming DataFrame/Dataset is written to a streaming sink.
+ * - `OutputMode.Append()`: only the new rows in the streaming DataFrame/Dataset will be
+ * written to the sink
+ * - `OutputMode.Complete()`: all the rows in the streaming DataFrame/Dataset will be written
+ * to the sink every time these is some updates
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def outputMode(outputMode: OutputMode): DataStreamWriter[T] = {
+ this.outputMode = outputMode
+ this
+ }
+
+
+ /**
+ * :: Experimental ::
+ * Specifies how data of a streaming DataFrame/Dataset is written to a streaming sink.
+ * - `append`: only the new rows in the streaming DataFrame/Dataset will be written to
+ * the sink
+ * - `complete`: all the rows in the streaming DataFrame/Dataset will be written to the sink
+ * every time these is some updates
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def outputMode(outputMode: String): DataStreamWriter[T] = {
--- End diff --
should this be shortened to just `mode`?
---
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 issue #13653: [SPARK-15933][SQL][STREAMING] Refactored DF reader-write...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/13653
**[Test build #60461 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/60461/consoleFull)** for PR 13653 at commit [`bbfff70`](https://github.com/apache/spark/commit/bbfff70b0bac8b2c33ab1eb736405759bb391633).
---
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 #13653: [SPARK-15933][SQL][STREAMING] Refactored DF reade...
Posted by tdas <gi...@git.apache.org>.
Github user tdas commented on a diff in the pull request:
https://github.com/apache/spark/pull/13653#discussion_r67082435
--- Diff: python/pyspark/sql/readwriter.py ---
@@ -905,6 +764,503 @@ def jdbc(self, url, table, mode=None, properties=None):
self._jwrite.mode(mode).jdbc(url, table, jprop)
+class DataStreamReader(object):
--- End diff --
I will fix this in the follow up pr #13673
---
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 #13653: [SPARK-15933][SQL][STREAMING] Refactored DF reade...
Posted by tdas <gi...@git.apache.org>.
Github user tdas commented on a diff in the pull request:
https://github.com/apache/spark/pull/13653#discussion_r67058676
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/streaming/DataStreamWriter.scala ---
@@ -0,0 +1,401 @@
+/*
+ * 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.streaming
+
+import scala.collection.JavaConverters._
+
+import org.apache.spark.annotation.Experimental
+import org.apache.spark.sql.{AnalysisException, DataFrame, Dataset, ForeachWriter}
+import org.apache.spark.sql.execution.datasources.DataSource
+import org.apache.spark.sql.execution.streaming.{ForeachSink, MemoryPlan, MemorySink}
+
+/**
+ * :: Experimental ::
+ * Interface used to write a streaming [[Dataset]] to external storage systems (e.g. file systems,
+ * key-value stores, etc). Use [[Dataset.write]] to access this.
+ *
+ * @since 2.0.0
+ */
+@Experimental
+final class DataStreamWriter[T] private[sql](ds: Dataset[T]) {
+
+ private val df = ds.toDF()
+
+ /**
+ * :: Experimental ::
+ * Specifies how data of a streaming DataFrame/Dataset is written to a streaming sink.
+ * - `OutputMode.Append()`: only the new rows in the streaming DataFrame/Dataset will be
+ * written to the sink
+ * - `OutputMode.Complete()`: all the rows in the streaming DataFrame/Dataset will be written
+ * to the sink every time these is some updates
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def outputMode(outputMode: OutputMode): DataStreamWriter[T] = {
+ this.outputMode = outputMode
+ this
+ }
+
+
+ /**
+ * :: Experimental ::
+ * Specifies how data of a streaming DataFrame/Dataset is written to a streaming sink.
+ * - `append`: only the new rows in the streaming DataFrame/Dataset will be written to
+ * the sink
+ * - `complete`: all the rows in the streaming DataFrame/Dataset will be written to the sink
+ * every time these is some updates
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def outputMode(outputMode: String): DataStreamWriter[T] = {
+ this.outputMode = outputMode.toLowerCase match {
+ case "append" =>
+ OutputMode.Append
+ case "complete" =>
+ OutputMode.Complete
+ case _ =>
+ throw new IllegalArgumentException(s"Unknown output mode $outputMode. " +
+ "Accepted output modes are 'append' and 'complete'")
+ }
+ this
+ }
+
+ /**
+ * :: Experimental ::
+ * Set the trigger for the stream query. The default value is `ProcessingTime(0)` and it will run
+ * the query as fast as possible.
+ *
+ * Scala Example:
+ * {{{
+ * df.write.trigger(ProcessingTime("10 seconds"))
+ *
+ * import scala.concurrent.duration._
+ * df.write.trigger(ProcessingTime(10.seconds))
+ * }}}
+ *
+ * Java Example:
+ * {{{
+ * df.write.trigger(ProcessingTime.create("10 seconds"))
+ *
+ * import java.util.concurrent.TimeUnit
+ * df.write.trigger(ProcessingTime.create(10, TimeUnit.SECONDS))
+ * }}}
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def trigger(trigger: Trigger): DataStreamWriter[T] = {
+ this.trigger = trigger
+ this
+ }
+
+
+ /**
+ * :: Experimental ::
+ * Specifies the name of the [[ContinuousQuery]] that can be started with `startStream()`.
+ * This name must be unique among all the currently active queries in the associated SQLContext.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def queryName(queryName: String): DataStreamWriter[T] = {
+ this.extraOptions += ("queryName" -> queryName)
+ this
+ }
+
+ /**
+ * :: Experimental ::
+ * Specifies the underlying output data source. Built-in options include "parquet", "json", etc.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def format(source: String): DataStreamWriter[T] = {
+ this.source = source
+ this
+ }
+
+ /**
+ * Partitions the output by the given columns on the file system. If specified, the output is
+ * laid out on the file system similar to Hive's partitioning scheme. As an example, when we
+ * partition a dataset by year and then month, the directory layout would look like:
+ *
+ * - year=2016/month=01/
+ * - year=2016/month=02/
+ *
+ * Partitioning is one of the most widely used techniques to optimize physical data layout.
+ * It provides a coarse-grained index for skipping unnecessary data reads when queries have
+ * predicates on the partitioned columns. In order for partitioning to work well, the number
+ * of distinct values in each column should typically be less than tens of thousands.
+ *
+ * This was initially applicable for Parquet but in 1.5+ covers JSON, text, ORC and avro as well.
+ *
+ * @since 1.4.0
+ */
+ @scala.annotation.varargs
+ def partitionBy(colNames: String*): DataStreamWriter[T] = {
+ this.partitioningColumns = Option(colNames)
+ this
+ }
+
+ /**
+ * :: Experimental ::
+ * Adds an output option for the underlying data source.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def option(key: String, value: String): DataStreamWriter[T] = {
+ this.extraOptions += (key -> value)
+ this
+ }
+
+ /**
+ * :: Experimental ::
+ * Adds an output option for the underlying data source.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def option(key: String, value: Boolean): DataStreamWriter[T] = option(key, value.toString)
+
+ /**
+ * :: Experimental ::
+ * Adds an output option for the underlying data source.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def option(key: String, value: Long): DataStreamWriter[T] = option(key, value.toString)
+
+ /**
+ * :: Experimental ::
+ * Adds an output option for the underlying data source.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def option(key: String, value: Double): DataStreamWriter[T] = option(key, value.toString)
+
+ /**
+ * :: Experimental ::
+ * (Scala-specific) Adds output options for the underlying data source.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def options(options: scala.collection.Map[String, String]): DataStreamWriter[T] = {
+ this.extraOptions ++= options
+ this
+ }
+
+ /**
+ * :: Experimental ::
+ * Adds output options for the underlying data source.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def options(options: java.util.Map[String, String]): DataStreamWriter[T] = {
+ this.options(options.asScala)
+ this
+ }
+
+
+ /**
+ * :: Experimental ::
+ * Saves the content of the streaming [[DataFrame]] in Parquet format at the specified path.
+ * This is equivalent to:
+ * {{{
+ * format("parquet").save(path)
--- End diff --
I am removing the parquet() method for now. Its nice to have a single method that starts the query - start()
---
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 issue #13653: [SPARK-15933][SQL][STREAMING] Refactored DF reader-write...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/13653
**[Test build #60452 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/60452/consoleFull)** for PR 13653 at commit [`a59498b`](https://github.com/apache/spark/commit/a59498bbe86609bc206de9b229052f35071049cb).
---
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 issue #13653: [SPARK-15933][SQL][STREAMING] Refactored DF reader-write...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/13653
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 issue #13653: [SPARK-15933][SQL][STREAMING] Refactored DF reader-write...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/13653
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/60461/
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 issue #13653: [SPARK-15933][SQL][STREAMING] Refactored DF reader-write...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/13653
**[Test build #60461 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/60461/consoleFull)** for PR 13653 at commit [`bbfff70`](https://github.com/apache/spark/commit/bbfff70b0bac8b2c33ab1eb736405759bb391633).
* 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 #13653: [SPARK-15933][SQL][STREAMING] Refactored DF reade...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/spark/pull/13653
---
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 #13653: [SPARK-15933][SQL][STREAMING] Refactored DF reade...
Posted by tdas <gi...@git.apache.org>.
Github user tdas commented on a diff in the pull request:
https://github.com/apache/spark/pull/13653#discussion_r67086340
--- Diff: sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingAggregationSuite.scala ---
@@ -40,6 +40,8 @@ class StreamingAggregationSuite extends StreamTest with BeforeAndAfterAll {
import testImplicits._
+
--- End diff --
Done in #13673
---
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 issue #13653: [SPARK-15933][SQL][STREAMING] Refactored DF reader-write...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/13653
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 #13653: [SPARK-15933][SQL][STREAMING] Refactored DF reade...
Posted by tdas <gi...@git.apache.org>.
Github user tdas commented on a diff in the pull request:
https://github.com/apache/spark/pull/13653#discussion_r66887929
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/streaming/DataStreamReader.scala ---
@@ -0,0 +1,288 @@
+/*
+ * 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.streaming
+
+import scala.collection.JavaConverters._
+
+import org.apache.spark.annotation.Experimental
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.{DataFrame, Dataset, SparkSession}
+import org.apache.spark.sql.execution.datasources.DataSource
+import org.apache.spark.sql.execution.streaming.StreamingRelation
+import org.apache.spark.sql.types.StructType
+
+@Experimental
+final class DataStreamReader private[sql](sparkSession: SparkSession) extends Logging {
+ /**
+ * :: Experimental ::
+ * Specifies the input data source format.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def format(source: String): DataStreamReader = {
+ this.source = source
+ this
+ }
+
+ /**
+ * :: Experimental ::
+ * Specifies the input schema. Some data sources (e.g. JSON) can infer the input schema
+ * automatically from data. By specifying the schema here, the underlying data source can
+ * skip the schema inference step, and thus speed up data loading.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def schema(schema: StructType): DataStreamReader = {
+ this.userSpecifiedSchema = Option(schema)
+ this
+ }
+
+ /**
+ * :: Experimental ::
+ * Adds an input option for the underlying data source.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def option(key: String, value: String): DataStreamReader = {
+ this.extraOptions += (key -> value)
+ this
+ }
+
+ /**
+ * :: Experimental ::
+ * Adds an input option for the underlying data source.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def option(key: String, value: Boolean): DataStreamReader = option(key, value.toString)
+
+ /**
+ * :: Experimental ::
+ * Adds an input option for the underlying data source.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def option(key: String, value: Long): DataStreamReader = option(key, value.toString)
+
+ /**
+ * :: Experimental ::
+ * Adds an input option for the underlying data source.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def option(key: String, value: Double): DataStreamReader = option(key, value.toString)
+
+ /**
+ * :: Experimental ::
+ * (Scala-specific) Adds input options for the underlying data source.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def options(options: scala.collection.Map[String, String]): DataStreamReader = {
+ this.extraOptions ++= options
+ this
+ }
+
+ /**
+ * :: Experimental ::
+ * Adds input options for the underlying data source.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def options(options: java.util.Map[String, String]): DataStreamReader = {
+ this.options(options.asScala)
+ this
+ }
+
+
+ /**
+ * :: Experimental ::
+ * Loads input data stream in as a [[DataFrame]], for data streams that don't require a path
+ * (e.g. external key-value stores).
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def load(): DataFrame = {
--- End diff --
@marmbrus @rxin Should this be `load()`? In the general case,
```
sdf.readStream
.format("myFormat")
.load("stringIdentifier")
```
---
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 #13653: [SPARK-15933][SQL][STREAMING] Refactored DF reade...
Posted by tdas <gi...@git.apache.org>.
Github user tdas commented on a diff in the pull request:
https://github.com/apache/spark/pull/13653#discussion_r66888017
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/streaming/DataStreamWriter.scala ---
@@ -0,0 +1,401 @@
+/*
+ * 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.streaming
+
+import scala.collection.JavaConverters._
+
+import org.apache.spark.annotation.Experimental
+import org.apache.spark.sql.{AnalysisException, DataFrame, Dataset, ForeachWriter}
+import org.apache.spark.sql.execution.datasources.DataSource
+import org.apache.spark.sql.execution.streaming.{ForeachSink, MemoryPlan, MemorySink}
+
+/**
+ * :: Experimental ::
+ * Interface used to write a streaming [[Dataset]] to external storage systems (e.g. file systems,
+ * key-value stores, etc). Use [[Dataset.write]] to access this.
+ *
+ * @since 2.0.0
+ */
+@Experimental
+final class DataStreamWriter[T] private[sql](ds: Dataset[T]) {
+
+ private val df = ds.toDF()
+
+ /**
+ * :: Experimental ::
+ * Specifies how data of a streaming DataFrame/Dataset is written to a streaming sink.
+ * - `OutputMode.Append()`: only the new rows in the streaming DataFrame/Dataset will be
+ * written to the sink
+ * - `OutputMode.Complete()`: all the rows in the streaming DataFrame/Dataset will be written
+ * to the sink every time these is some updates
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def outputMode(outputMode: OutputMode): DataStreamWriter[T] = {
+ this.outputMode = outputMode
+ this
+ }
+
+
+ /**
+ * :: Experimental ::
+ * Specifies how data of a streaming DataFrame/Dataset is written to a streaming sink.
+ * - `append`: only the new rows in the streaming DataFrame/Dataset will be written to
+ * the sink
+ * - `complete`: all the rows in the streaming DataFrame/Dataset will be written to the sink
+ * every time these is some updates
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def outputMode(outputMode: String): DataStreamWriter[T] = {
+ this.outputMode = outputMode.toLowerCase match {
+ case "append" =>
+ OutputMode.Append
+ case "complete" =>
+ OutputMode.Complete
+ case _ =>
+ throw new IllegalArgumentException(s"Unknown output mode $outputMode. " +
+ "Accepted output modes are 'append' and 'complete'")
+ }
+ this
+ }
+
+ /**
+ * :: Experimental ::
+ * Set the trigger for the stream query. The default value is `ProcessingTime(0)` and it will run
+ * the query as fast as possible.
+ *
+ * Scala Example:
+ * {{{
+ * df.write.trigger(ProcessingTime("10 seconds"))
--- End diff --
update these examples.
---
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 #13653: [SPARK-15933][SQL][STREAMING] Refactored DF reade...
Posted by marmbrus <gi...@git.apache.org>.
Github user marmbrus commented on a diff in the pull request:
https://github.com/apache/spark/pull/13653#discussion_r67022595
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/streaming/DataStreamReader.scala ---
@@ -0,0 +1,288 @@
+/*
+ * 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.streaming
+
+import scala.collection.JavaConverters._
+
+import org.apache.spark.annotation.Experimental
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.{DataFrame, Dataset, SparkSession}
+import org.apache.spark.sql.execution.datasources.DataSource
+import org.apache.spark.sql.execution.streaming.StreamingRelation
+import org.apache.spark.sql.types.StructType
+
+@Experimental
+final class DataStreamReader private[sql](sparkSession: SparkSession) extends Logging {
+ /**
+ * :: Experimental ::
+ * Specifies the input data source format.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def format(source: String): DataStreamReader = {
+ this.source = source
+ this
+ }
+
+ /**
+ * :: Experimental ::
+ * Specifies the input schema. Some data sources (e.g. JSON) can infer the input schema
+ * automatically from data. By specifying the schema here, the underlying data source can
+ * skip the schema inference step, and thus speed up data loading.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def schema(schema: StructType): DataStreamReader = {
+ this.userSpecifiedSchema = Option(schema)
+ this
+ }
+
+ /**
+ * :: Experimental ::
+ * Adds an input option for the underlying data source.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def option(key: String, value: String): DataStreamReader = {
+ this.extraOptions += (key -> value)
+ this
+ }
+
+ /**
+ * :: Experimental ::
+ * Adds an input option for the underlying data source.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def option(key: String, value: Boolean): DataStreamReader = option(key, value.toString)
+
+ /**
+ * :: Experimental ::
+ * Adds an input option for the underlying data source.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def option(key: String, value: Long): DataStreamReader = option(key, value.toString)
+
+ /**
+ * :: Experimental ::
+ * Adds an input option for the underlying data source.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def option(key: String, value: Double): DataStreamReader = option(key, value.toString)
+
+ /**
+ * :: Experimental ::
+ * (Scala-specific) Adds input options for the underlying data source.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def options(options: scala.collection.Map[String, String]): DataStreamReader = {
+ this.extraOptions ++= options
+ this
+ }
+
+ /**
+ * :: Experimental ::
+ * Adds input options for the underlying data source.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def options(options: java.util.Map[String, String]): DataStreamReader = {
+ this.options(options.asScala)
+ this
+ }
+
+
+ /**
+ * :: Experimental ::
+ * Loads input data stream in as a [[DataFrame]], for data streams that don't require a path
+ * (e.g. external key-value stores).
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def load(): DataFrame = {
--- End diff --
Seems best to keep it consistent.
---
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 #13653: [SPARK-15933][SQL][STREAMING] Refactored DF reade...
Posted by marmbrus <gi...@git.apache.org>.
Github user marmbrus commented on a diff in the pull request:
https://github.com/apache/spark/pull/13653#discussion_r67023216
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/streaming/DataStreamWriter.scala ---
@@ -0,0 +1,401 @@
+/*
+ * 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.streaming
+
+import scala.collection.JavaConverters._
+
+import org.apache.spark.annotation.Experimental
+import org.apache.spark.sql.{AnalysisException, DataFrame, Dataset, ForeachWriter}
+import org.apache.spark.sql.execution.datasources.DataSource
+import org.apache.spark.sql.execution.streaming.{ForeachSink, MemoryPlan, MemorySink}
+
+/**
+ * :: Experimental ::
+ * Interface used to write a streaming [[Dataset]] to external storage systems (e.g. file systems,
+ * key-value stores, etc). Use [[Dataset.write]] to access this.
+ *
+ * @since 2.0.0
+ */
+@Experimental
+final class DataStreamWriter[T] private[sql](ds: Dataset[T]) {
+
+ private val df = ds.toDF()
+
+ /**
+ * :: Experimental ::
+ * Specifies how data of a streaming DataFrame/Dataset is written to a streaming sink.
+ * - `OutputMode.Append()`: only the new rows in the streaming DataFrame/Dataset will be
+ * written to the sink
+ * - `OutputMode.Complete()`: all the rows in the streaming DataFrame/Dataset will be written
+ * to the sink every time these is some updates
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def outputMode(outputMode: OutputMode): DataStreamWriter[T] = {
+ this.outputMode = outputMode
+ this
+ }
+
+
+ /**
+ * :: Experimental ::
+ * Specifies how data of a streaming DataFrame/Dataset is written to a streaming sink.
+ * - `append`: only the new rows in the streaming DataFrame/Dataset will be written to
+ * the sink
+ * - `complete`: all the rows in the streaming DataFrame/Dataset will be written to the sink
+ * every time these is some updates
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def outputMode(outputMode: String): DataStreamWriter[T] = {
+ this.outputMode = outputMode.toLowerCase match {
+ case "append" =>
+ OutputMode.Append
+ case "complete" =>
+ OutputMode.Complete
+ case _ =>
+ throw new IllegalArgumentException(s"Unknown output mode $outputMode. " +
+ "Accepted output modes are 'append' and 'complete'")
+ }
+ this
+ }
+
+ /**
+ * :: Experimental ::
+ * Set the trigger for the stream query. The default value is `ProcessingTime(0)` and it will run
+ * the query as fast as possible.
+ *
+ * Scala Example:
+ * {{{
+ * df.write.trigger(ProcessingTime("10 seconds"))
+ *
+ * import scala.concurrent.duration._
+ * df.write.trigger(ProcessingTime(10.seconds))
+ * }}}
+ *
+ * Java Example:
+ * {{{
+ * df.write.trigger(ProcessingTime.create("10 seconds"))
+ *
+ * import java.util.concurrent.TimeUnit
+ * df.write.trigger(ProcessingTime.create(10, TimeUnit.SECONDS))
+ * }}}
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def trigger(trigger: Trigger): DataStreamWriter[T] = {
+ this.trigger = trigger
+ this
+ }
+
+
+ /**
+ * :: Experimental ::
+ * Specifies the name of the [[ContinuousQuery]] that can be started with `startStream()`.
+ * This name must be unique among all the currently active queries in the associated SQLContext.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def queryName(queryName: String): DataStreamWriter[T] = {
+ this.extraOptions += ("queryName" -> queryName)
+ this
+ }
+
+ /**
+ * :: Experimental ::
+ * Specifies the underlying output data source. Built-in options include "parquet", "json", etc.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def format(source: String): DataStreamWriter[T] = {
+ this.source = source
+ this
+ }
+
+ /**
+ * Partitions the output by the given columns on the file system. If specified, the output is
+ * laid out on the file system similar to Hive's partitioning scheme. As an example, when we
+ * partition a dataset by year and then month, the directory layout would look like:
+ *
+ * - year=2016/month=01/
+ * - year=2016/month=02/
+ *
+ * Partitioning is one of the most widely used techniques to optimize physical data layout.
+ * It provides a coarse-grained index for skipping unnecessary data reads when queries have
+ * predicates on the partitioned columns. In order for partitioning to work well, the number
+ * of distinct values in each column should typically be less than tens of thousands.
+ *
+ * This was initially applicable for Parquet but in 1.5+ covers JSON, text, ORC and avro as well.
+ *
+ * @since 1.4.0
+ */
+ @scala.annotation.varargs
+ def partitionBy(colNames: String*): DataStreamWriter[T] = {
+ this.partitioningColumns = Option(colNames)
+ this
+ }
+
+ /**
+ * :: Experimental ::
+ * Adds an output option for the underlying data source.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def option(key: String, value: String): DataStreamWriter[T] = {
+ this.extraOptions += (key -> value)
+ this
+ }
+
+ /**
+ * :: Experimental ::
+ * Adds an output option for the underlying data source.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def option(key: String, value: Boolean): DataStreamWriter[T] = option(key, value.toString)
+
+ /**
+ * :: Experimental ::
+ * Adds an output option for the underlying data source.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def option(key: String, value: Long): DataStreamWriter[T] = option(key, value.toString)
+
+ /**
+ * :: Experimental ::
+ * Adds an output option for the underlying data source.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def option(key: String, value: Double): DataStreamWriter[T] = option(key, value.toString)
+
+ /**
+ * :: Experimental ::
+ * (Scala-specific) Adds output options for the underlying data source.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def options(options: scala.collection.Map[String, String]): DataStreamWriter[T] = {
+ this.extraOptions ++= options
+ this
+ }
+
+ /**
+ * :: Experimental ::
+ * Adds output options for the underlying data source.
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def options(options: java.util.Map[String, String]): DataStreamWriter[T] = {
+ this.options(options.asScala)
+ this
+ }
+
+
+ /**
+ * :: Experimental ::
+ * Saves the content of the streaming [[DataFrame]] in Parquet format at the specified path.
+ * This is equivalent to:
+ * {{{
+ * format("parquet").save(path)
--- End diff --
this should probably start with `df.writeStream.`
---
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 issue #13653: [SPARK-15933][SQL][STREAMING] Refactored DF reader-write...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/13653
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 #13653: [SPARK-15933][SQL][STREAMING] Refactored DF reade...
Posted by marmbrus <gi...@git.apache.org>.
Github user marmbrus commented on a diff in the pull request:
https://github.com/apache/spark/pull/13653#discussion_r67023075
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/streaming/DataStreamWriter.scala ---
@@ -0,0 +1,401 @@
+/*
+ * 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.streaming
+
+import scala.collection.JavaConverters._
+
+import org.apache.spark.annotation.Experimental
+import org.apache.spark.sql.{AnalysisException, DataFrame, Dataset, ForeachWriter}
+import org.apache.spark.sql.execution.datasources.DataSource
+import org.apache.spark.sql.execution.streaming.{ForeachSink, MemoryPlan, MemorySink}
+
+/**
+ * :: Experimental ::
+ * Interface used to write a streaming [[Dataset]] to external storage systems (e.g. file systems,
+ * key-value stores, etc). Use [[Dataset.write]] to access this.
+ *
+ * @since 2.0.0
+ */
+@Experimental
+final class DataStreamWriter[T] private[sql](ds: Dataset[T]) {
+
+ private val df = ds.toDF()
+
+ /**
+ * :: Experimental ::
+ * Specifies how data of a streaming DataFrame/Dataset is written to a streaming sink.
+ * - `OutputMode.Append()`: only the new rows in the streaming DataFrame/Dataset will be
+ * written to the sink
+ * - `OutputMode.Complete()`: all the rows in the streaming DataFrame/Dataset will be written
+ * to the sink every time these is some updates
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def outputMode(outputMode: OutputMode): DataStreamWriter[T] = {
+ this.outputMode = outputMode
+ this
+ }
+
+
+ /**
+ * :: Experimental ::
+ * Specifies how data of a streaming DataFrame/Dataset is written to a streaming sink.
+ * - `append`: only the new rows in the streaming DataFrame/Dataset will be written to
+ * the sink
+ * - `complete`: all the rows in the streaming DataFrame/Dataset will be written to the sink
+ * every time these is some updates
+ *
+ * @since 2.0.0
+ */
+ @Experimental
+ def outputMode(outputMode: String): DataStreamWriter[T] = {
--- End diff --
Again, I'd probably go for consistency unless we think that we will need save modes in streaming (i'm not really sure how that would work).
---
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