2016-04-20 13:32:01 -04:00
|
|
|
#
|
|
|
|
# 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.
|
|
|
|
#
|
|
|
|
|
2016-06-14 05:12:29 -04:00
|
|
|
import sys
|
[SPARK-18516][SQL] Split state and progress in streaming
This PR separates the status of a `StreamingQuery` into two separate APIs:
- `status` - describes the status of a `StreamingQuery` at this moment, including what phase of processing is currently happening and if data is available.
- `recentProgress` - an array of statistics about the most recent microbatches that have executed.
A recent progress contains the following information:
```
{
"id" : "2be8670a-fce1-4859-a530-748f29553bb6",
"name" : "query-29",
"timestamp" : 1479705392724,
"inputRowsPerSecond" : 230.76923076923077,
"processedRowsPerSecond" : 10.869565217391303,
"durationMs" : {
"triggerExecution" : 276,
"queryPlanning" : 3,
"getBatch" : 5,
"getOffset" : 3,
"addBatch" : 234,
"walCommit" : 30
},
"currentWatermark" : 0,
"stateOperators" : [ ],
"sources" : [ {
"description" : "KafkaSource[Subscribe[topic-14]]",
"startOffset" : {
"topic-14" : {
"2" : 0,
"4" : 1,
"1" : 0,
"3" : 0,
"0" : 0
}
},
"endOffset" : {
"topic-14" : {
"2" : 1,
"4" : 2,
"1" : 0,
"3" : 0,
"0" : 1
}
},
"numRecords" : 3,
"inputRowsPerSecond" : 230.76923076923077,
"processedRowsPerSecond" : 10.869565217391303
} ]
}
```
Additionally, in order to make it possible to correlate progress updates across restarts, we change the `id` field from an integer that is unique with in the JVM to a `UUID` that is globally unique.
Author: Tathagata Das <tathagata.das1565@gmail.com>
Author: Michael Armbrust <michael@databricks.com>
Closes #15954 from marmbrus/queryProgress.
2016-11-29 20:24:17 -05:00
|
|
|
import json
|
|
|
|
|
2016-06-14 05:12:29 -04:00
|
|
|
if sys.version >= '3':
|
|
|
|
intlike = int
|
2016-06-29 01:07:11 -04:00
|
|
|
basestring = unicode = str
|
2016-06-14 05:12:29 -04:00
|
|
|
else:
|
|
|
|
intlike = (int, long)
|
|
|
|
|
2016-06-29 01:07:11 -04:00
|
|
|
from pyspark import since, keyword_only
|
2016-04-28 18:22:28 -04:00
|
|
|
from pyspark.rdd import ignore_unicode_prefix
|
2016-12-15 17:26:54 -05:00
|
|
|
from pyspark.sql.column import _to_seq
|
2016-06-29 01:07:11 -04:00
|
|
|
from pyspark.sql.readwriter import OptionUtils, to_str
|
|
|
|
from pyspark.sql.types import *
|
2016-12-05 14:36:11 -05:00
|
|
|
from pyspark.sql.utils import StreamingQueryException
|
2016-04-20 13:32:01 -04:00
|
|
|
|
2016-06-29 01:07:11 -04:00
|
|
|
__all__ = ["StreamingQuery", "StreamingQueryManager", "DataStreamReader", "DataStreamWriter"]
|
2016-04-20 13:32:01 -04:00
|
|
|
|
|
|
|
|
2016-06-15 13:46:02 -04:00
|
|
|
class StreamingQuery(object):
|
2016-04-20 13:32:01 -04:00
|
|
|
"""
|
|
|
|
A handle to a query that is executing continuously in the background as new data arrives.
|
|
|
|
All these methods are thread-safe.
|
|
|
|
|
2017-05-26 16:33:23 -04:00
|
|
|
.. note:: Evolving
|
2016-04-20 13:32:01 -04:00
|
|
|
|
|
|
|
.. versionadded:: 2.0
|
|
|
|
"""
|
|
|
|
|
2016-06-15 13:46:02 -04:00
|
|
|
def __init__(self, jsq):
|
|
|
|
self._jsq = jsq
|
2016-04-20 13:32:01 -04:00
|
|
|
|
2016-06-14 05:12:29 -04:00
|
|
|
@property
|
|
|
|
@since(2.0)
|
|
|
|
def id(self):
|
2016-12-05 21:17:38 -05:00
|
|
|
"""Returns the unique id of this query that persists across restarts from checkpoint data.
|
|
|
|
That is, this id is generated when a query is started for the first time, and
|
|
|
|
will be the same every time it is restarted from checkpoint data.
|
|
|
|
There can only be one query with the same id active in a Spark cluster.
|
|
|
|
Also see, `runId`.
|
2016-06-14 05:12:29 -04:00
|
|
|
"""
|
[SPARK-18516][SQL] Split state and progress in streaming
This PR separates the status of a `StreamingQuery` into two separate APIs:
- `status` - describes the status of a `StreamingQuery` at this moment, including what phase of processing is currently happening and if data is available.
- `recentProgress` - an array of statistics about the most recent microbatches that have executed.
A recent progress contains the following information:
```
{
"id" : "2be8670a-fce1-4859-a530-748f29553bb6",
"name" : "query-29",
"timestamp" : 1479705392724,
"inputRowsPerSecond" : 230.76923076923077,
"processedRowsPerSecond" : 10.869565217391303,
"durationMs" : {
"triggerExecution" : 276,
"queryPlanning" : 3,
"getBatch" : 5,
"getOffset" : 3,
"addBatch" : 234,
"walCommit" : 30
},
"currentWatermark" : 0,
"stateOperators" : [ ],
"sources" : [ {
"description" : "KafkaSource[Subscribe[topic-14]]",
"startOffset" : {
"topic-14" : {
"2" : 0,
"4" : 1,
"1" : 0,
"3" : 0,
"0" : 0
}
},
"endOffset" : {
"topic-14" : {
"2" : 1,
"4" : 2,
"1" : 0,
"3" : 0,
"0" : 1
}
},
"numRecords" : 3,
"inputRowsPerSecond" : 230.76923076923077,
"processedRowsPerSecond" : 10.869565217391303
} ]
}
```
Additionally, in order to make it possible to correlate progress updates across restarts, we change the `id` field from an integer that is unique with in the JVM to a `UUID` that is globally unique.
Author: Tathagata Das <tathagata.das1565@gmail.com>
Author: Michael Armbrust <michael@databricks.com>
Closes #15954 from marmbrus/queryProgress.
2016-11-29 20:24:17 -05:00
|
|
|
return self._jsq.id().toString()
|
2016-06-14 05:12:29 -04:00
|
|
|
|
2016-12-05 21:17:38 -05:00
|
|
|
@property
|
|
|
|
@since(2.1)
|
|
|
|
def runId(self):
|
|
|
|
"""Returns the unique id of this query that does not persist across restarts. That is, every
|
|
|
|
query that is started (or restarted from checkpoint) will have a different runId.
|
|
|
|
"""
|
|
|
|
return self._jsq.runId().toString()
|
|
|
|
|
2016-04-20 13:32:01 -04:00
|
|
|
@property
|
|
|
|
@since(2.0)
|
|
|
|
def name(self):
|
2016-12-05 21:17:38 -05:00
|
|
|
"""Returns the user-specified name of the query, or null if not specified.
|
|
|
|
This name can be specified in the `org.apache.spark.sql.streaming.DataStreamWriter`
|
|
|
|
as `dataframe.writeStream.queryName("query").start()`.
|
|
|
|
This name, if set, must be unique across all active queries.
|
2016-04-20 13:32:01 -04:00
|
|
|
"""
|
2016-06-15 13:46:02 -04:00
|
|
|
return self._jsq.name()
|
2016-04-20 13:32:01 -04:00
|
|
|
|
|
|
|
@property
|
|
|
|
@since(2.0)
|
|
|
|
def isActive(self):
|
2016-06-15 13:46:02 -04:00
|
|
|
"""Whether this streaming query is currently active or not.
|
2016-04-20 13:32:01 -04:00
|
|
|
"""
|
2016-06-15 13:46:02 -04:00
|
|
|
return self._jsq.isActive()
|
2016-04-20 13:32:01 -04:00
|
|
|
|
|
|
|
@since(2.0)
|
2016-04-28 18:22:28 -04:00
|
|
|
def awaitTermination(self, timeout=None):
|
2016-04-20 13:32:01 -04:00
|
|
|
"""Waits for the termination of `this` query, either by :func:`query.stop()` or by an
|
|
|
|
exception. If the query has terminated with an exception, then the exception will be thrown.
|
2016-04-28 18:22:28 -04:00
|
|
|
If `timeout` is set, it returns whether the query has terminated or not within the
|
|
|
|
`timeout` seconds.
|
2016-04-20 13:32:01 -04:00
|
|
|
|
|
|
|
If the query has terminated, then all subsequent calls to this method will either return
|
|
|
|
immediately (if the query was terminated by :func:`stop()`), or throw the exception
|
|
|
|
immediately (if the query has terminated with exception).
|
|
|
|
|
2016-06-15 13:46:02 -04:00
|
|
|
throws :class:`StreamingQueryException`, if `this` query has terminated with an exception
|
2016-04-20 13:32:01 -04:00
|
|
|
"""
|
2016-04-28 18:22:28 -04:00
|
|
|
if timeout is not None:
|
|
|
|
if not isinstance(timeout, (int, float)) or timeout < 0:
|
|
|
|
raise ValueError("timeout must be a positive integer or float. Got %s" % timeout)
|
2016-06-15 13:46:02 -04:00
|
|
|
return self._jsq.awaitTermination(int(timeout * 1000))
|
2016-04-20 13:32:01 -04:00
|
|
|
else:
|
2016-06-15 13:46:02 -04:00
|
|
|
return self._jsq.awaitTermination()
|
2016-04-20 13:32:01 -04:00
|
|
|
|
2016-11-30 02:08:56 -05:00
|
|
|
@property
|
|
|
|
@since(2.1)
|
|
|
|
def status(self):
|
|
|
|
"""
|
|
|
|
Returns the current status of the query.
|
|
|
|
"""
|
|
|
|
return json.loads(self._jsq.status().json())
|
|
|
|
|
[SPARK-18516][SQL] Split state and progress in streaming
This PR separates the status of a `StreamingQuery` into two separate APIs:
- `status` - describes the status of a `StreamingQuery` at this moment, including what phase of processing is currently happening and if data is available.
- `recentProgress` - an array of statistics about the most recent microbatches that have executed.
A recent progress contains the following information:
```
{
"id" : "2be8670a-fce1-4859-a530-748f29553bb6",
"name" : "query-29",
"timestamp" : 1479705392724,
"inputRowsPerSecond" : 230.76923076923077,
"processedRowsPerSecond" : 10.869565217391303,
"durationMs" : {
"triggerExecution" : 276,
"queryPlanning" : 3,
"getBatch" : 5,
"getOffset" : 3,
"addBatch" : 234,
"walCommit" : 30
},
"currentWatermark" : 0,
"stateOperators" : [ ],
"sources" : [ {
"description" : "KafkaSource[Subscribe[topic-14]]",
"startOffset" : {
"topic-14" : {
"2" : 0,
"4" : 1,
"1" : 0,
"3" : 0,
"0" : 0
}
},
"endOffset" : {
"topic-14" : {
"2" : 1,
"4" : 2,
"1" : 0,
"3" : 0,
"0" : 1
}
},
"numRecords" : 3,
"inputRowsPerSecond" : 230.76923076923077,
"processedRowsPerSecond" : 10.869565217391303
} ]
}
```
Additionally, in order to make it possible to correlate progress updates across restarts, we change the `id` field from an integer that is unique with in the JVM to a `UUID` that is globally unique.
Author: Tathagata Das <tathagata.das1565@gmail.com>
Author: Michael Armbrust <michael@databricks.com>
Closes #15954 from marmbrus/queryProgress.
2016-11-29 20:24:17 -05:00
|
|
|
@property
|
|
|
|
@since(2.1)
|
2016-12-07 18:36:29 -05:00
|
|
|
def recentProgress(self):
|
[SPARK-18516][SQL] Split state and progress in streaming
This PR separates the status of a `StreamingQuery` into two separate APIs:
- `status` - describes the status of a `StreamingQuery` at this moment, including what phase of processing is currently happening and if data is available.
- `recentProgress` - an array of statistics about the most recent microbatches that have executed.
A recent progress contains the following information:
```
{
"id" : "2be8670a-fce1-4859-a530-748f29553bb6",
"name" : "query-29",
"timestamp" : 1479705392724,
"inputRowsPerSecond" : 230.76923076923077,
"processedRowsPerSecond" : 10.869565217391303,
"durationMs" : {
"triggerExecution" : 276,
"queryPlanning" : 3,
"getBatch" : 5,
"getOffset" : 3,
"addBatch" : 234,
"walCommit" : 30
},
"currentWatermark" : 0,
"stateOperators" : [ ],
"sources" : [ {
"description" : "KafkaSource[Subscribe[topic-14]]",
"startOffset" : {
"topic-14" : {
"2" : 0,
"4" : 1,
"1" : 0,
"3" : 0,
"0" : 0
}
},
"endOffset" : {
"topic-14" : {
"2" : 1,
"4" : 2,
"1" : 0,
"3" : 0,
"0" : 1
}
},
"numRecords" : 3,
"inputRowsPerSecond" : 230.76923076923077,
"processedRowsPerSecond" : 10.869565217391303
} ]
}
```
Additionally, in order to make it possible to correlate progress updates across restarts, we change the `id` field from an integer that is unique with in the JVM to a `UUID` that is globally unique.
Author: Tathagata Das <tathagata.das1565@gmail.com>
Author: Michael Armbrust <michael@databricks.com>
Closes #15954 from marmbrus/queryProgress.
2016-11-29 20:24:17 -05:00
|
|
|
"""Returns an array of the most recent [[StreamingQueryProgress]] updates for this query.
|
|
|
|
The number of progress updates retained for each stream is configured by Spark session
|
2016-12-07 18:36:29 -05:00
|
|
|
configuration `spark.sql.streaming.numRecentProgressUpdates`.
|
[SPARK-18516][SQL] Split state and progress in streaming
This PR separates the status of a `StreamingQuery` into two separate APIs:
- `status` - describes the status of a `StreamingQuery` at this moment, including what phase of processing is currently happening and if data is available.
- `recentProgress` - an array of statistics about the most recent microbatches that have executed.
A recent progress contains the following information:
```
{
"id" : "2be8670a-fce1-4859-a530-748f29553bb6",
"name" : "query-29",
"timestamp" : 1479705392724,
"inputRowsPerSecond" : 230.76923076923077,
"processedRowsPerSecond" : 10.869565217391303,
"durationMs" : {
"triggerExecution" : 276,
"queryPlanning" : 3,
"getBatch" : 5,
"getOffset" : 3,
"addBatch" : 234,
"walCommit" : 30
},
"currentWatermark" : 0,
"stateOperators" : [ ],
"sources" : [ {
"description" : "KafkaSource[Subscribe[topic-14]]",
"startOffset" : {
"topic-14" : {
"2" : 0,
"4" : 1,
"1" : 0,
"3" : 0,
"0" : 0
}
},
"endOffset" : {
"topic-14" : {
"2" : 1,
"4" : 2,
"1" : 0,
"3" : 0,
"0" : 1
}
},
"numRecords" : 3,
"inputRowsPerSecond" : 230.76923076923077,
"processedRowsPerSecond" : 10.869565217391303
} ]
}
```
Additionally, in order to make it possible to correlate progress updates across restarts, we change the `id` field from an integer that is unique with in the JVM to a `UUID` that is globally unique.
Author: Tathagata Das <tathagata.das1565@gmail.com>
Author: Michael Armbrust <michael@databricks.com>
Closes #15954 from marmbrus/queryProgress.
2016-11-29 20:24:17 -05:00
|
|
|
"""
|
2016-12-07 18:36:29 -05:00
|
|
|
return [json.loads(p.json()) for p in self._jsq.recentProgress()]
|
[SPARK-18516][SQL] Split state and progress in streaming
This PR separates the status of a `StreamingQuery` into two separate APIs:
- `status` - describes the status of a `StreamingQuery` at this moment, including what phase of processing is currently happening and if data is available.
- `recentProgress` - an array of statistics about the most recent microbatches that have executed.
A recent progress contains the following information:
```
{
"id" : "2be8670a-fce1-4859-a530-748f29553bb6",
"name" : "query-29",
"timestamp" : 1479705392724,
"inputRowsPerSecond" : 230.76923076923077,
"processedRowsPerSecond" : 10.869565217391303,
"durationMs" : {
"triggerExecution" : 276,
"queryPlanning" : 3,
"getBatch" : 5,
"getOffset" : 3,
"addBatch" : 234,
"walCommit" : 30
},
"currentWatermark" : 0,
"stateOperators" : [ ],
"sources" : [ {
"description" : "KafkaSource[Subscribe[topic-14]]",
"startOffset" : {
"topic-14" : {
"2" : 0,
"4" : 1,
"1" : 0,
"3" : 0,
"0" : 0
}
},
"endOffset" : {
"topic-14" : {
"2" : 1,
"4" : 2,
"1" : 0,
"3" : 0,
"0" : 1
}
},
"numRecords" : 3,
"inputRowsPerSecond" : 230.76923076923077,
"processedRowsPerSecond" : 10.869565217391303
} ]
}
```
Additionally, in order to make it possible to correlate progress updates across restarts, we change the `id` field from an integer that is unique with in the JVM to a `UUID` that is globally unique.
Author: Tathagata Das <tathagata.das1565@gmail.com>
Author: Michael Armbrust <michael@databricks.com>
Closes #15954 from marmbrus/queryProgress.
2016-11-29 20:24:17 -05:00
|
|
|
|
|
|
|
@property
|
|
|
|
@since(2.1)
|
|
|
|
def lastProgress(self):
|
|
|
|
"""
|
2016-12-14 16:36:41 -05:00
|
|
|
Returns the most recent :class:`StreamingQueryProgress` update of this streaming query or
|
|
|
|
None if there were no progress updates
|
[SPARK-18516][SQL] Split state and progress in streaming
This PR separates the status of a `StreamingQuery` into two separate APIs:
- `status` - describes the status of a `StreamingQuery` at this moment, including what phase of processing is currently happening and if data is available.
- `recentProgress` - an array of statistics about the most recent microbatches that have executed.
A recent progress contains the following information:
```
{
"id" : "2be8670a-fce1-4859-a530-748f29553bb6",
"name" : "query-29",
"timestamp" : 1479705392724,
"inputRowsPerSecond" : 230.76923076923077,
"processedRowsPerSecond" : 10.869565217391303,
"durationMs" : {
"triggerExecution" : 276,
"queryPlanning" : 3,
"getBatch" : 5,
"getOffset" : 3,
"addBatch" : 234,
"walCommit" : 30
},
"currentWatermark" : 0,
"stateOperators" : [ ],
"sources" : [ {
"description" : "KafkaSource[Subscribe[topic-14]]",
"startOffset" : {
"topic-14" : {
"2" : 0,
"4" : 1,
"1" : 0,
"3" : 0,
"0" : 0
}
},
"endOffset" : {
"topic-14" : {
"2" : 1,
"4" : 2,
"1" : 0,
"3" : 0,
"0" : 1
}
},
"numRecords" : 3,
"inputRowsPerSecond" : 230.76923076923077,
"processedRowsPerSecond" : 10.869565217391303
} ]
}
```
Additionally, in order to make it possible to correlate progress updates across restarts, we change the `id` field from an integer that is unique with in the JVM to a `UUID` that is globally unique.
Author: Tathagata Das <tathagata.das1565@gmail.com>
Author: Michael Armbrust <michael@databricks.com>
Closes #15954 from marmbrus/queryProgress.
2016-11-29 20:24:17 -05:00
|
|
|
:return: a map
|
|
|
|
"""
|
2016-12-14 16:36:41 -05:00
|
|
|
lastProgress = self._jsq.lastProgress()
|
|
|
|
if lastProgress:
|
|
|
|
return json.loads(lastProgress.json())
|
|
|
|
else:
|
|
|
|
return None
|
[SPARK-18516][SQL] Split state and progress in streaming
This PR separates the status of a `StreamingQuery` into two separate APIs:
- `status` - describes the status of a `StreamingQuery` at this moment, including what phase of processing is currently happening and if data is available.
- `recentProgress` - an array of statistics about the most recent microbatches that have executed.
A recent progress contains the following information:
```
{
"id" : "2be8670a-fce1-4859-a530-748f29553bb6",
"name" : "query-29",
"timestamp" : 1479705392724,
"inputRowsPerSecond" : 230.76923076923077,
"processedRowsPerSecond" : 10.869565217391303,
"durationMs" : {
"triggerExecution" : 276,
"queryPlanning" : 3,
"getBatch" : 5,
"getOffset" : 3,
"addBatch" : 234,
"walCommit" : 30
},
"currentWatermark" : 0,
"stateOperators" : [ ],
"sources" : [ {
"description" : "KafkaSource[Subscribe[topic-14]]",
"startOffset" : {
"topic-14" : {
"2" : 0,
"4" : 1,
"1" : 0,
"3" : 0,
"0" : 0
}
},
"endOffset" : {
"topic-14" : {
"2" : 1,
"4" : 2,
"1" : 0,
"3" : 0,
"0" : 1
}
},
"numRecords" : 3,
"inputRowsPerSecond" : 230.76923076923077,
"processedRowsPerSecond" : 10.869565217391303
} ]
}
```
Additionally, in order to make it possible to correlate progress updates across restarts, we change the `id` field from an integer that is unique with in the JVM to a `UUID` that is globally unique.
Author: Tathagata Das <tathagata.das1565@gmail.com>
Author: Michael Armbrust <michael@databricks.com>
Closes #15954 from marmbrus/queryProgress.
2016-11-29 20:24:17 -05:00
|
|
|
|
2016-04-20 13:32:01 -04:00
|
|
|
@since(2.0)
|
|
|
|
def processAllAvailable(self):
|
2016-06-06 04:35:47 -04:00
|
|
|
"""Blocks until all available data in the source has been processed and committed to the
|
2016-11-22 06:40:18 -05:00
|
|
|
sink. This method is intended for testing.
|
|
|
|
|
|
|
|
.. note:: In the case of continually arriving data, this method may block forever.
|
|
|
|
Additionally, this method is only guaranteed to block until data that has been
|
|
|
|
synchronously appended data to a stream source prior to invocation.
|
|
|
|
(i.e. `getOffset` must immediately reflect the addition).
|
2016-04-20 13:32:01 -04:00
|
|
|
"""
|
2016-06-15 13:46:02 -04:00
|
|
|
return self._jsq.processAllAvailable()
|
2016-04-20 13:32:01 -04:00
|
|
|
|
|
|
|
@since(2.0)
|
|
|
|
def stop(self):
|
2016-06-15 13:46:02 -04:00
|
|
|
"""Stop this streaming query.
|
2016-04-20 13:32:01 -04:00
|
|
|
"""
|
2016-06-15 13:46:02 -04:00
|
|
|
self._jsq.stop()
|
2016-04-20 13:32:01 -04:00
|
|
|
|
2016-12-05 14:36:11 -05:00
|
|
|
@since(2.1)
|
|
|
|
def explain(self, extended=False):
|
|
|
|
"""Prints the (logical and physical) plans to the console for debugging purpose.
|
|
|
|
|
|
|
|
:param extended: boolean, default ``False``. If ``False``, prints only the physical plan.
|
|
|
|
|
|
|
|
>>> sq = sdf.writeStream.format('memory').queryName('query_explain').start()
|
|
|
|
>>> sq.processAllAvailable() # Wait a bit to generate the runtime plans.
|
|
|
|
>>> sq.explain()
|
|
|
|
== Physical Plan ==
|
|
|
|
...
|
|
|
|
>>> sq.explain(True)
|
|
|
|
== Parsed Logical Plan ==
|
|
|
|
...
|
|
|
|
== Analyzed Logical Plan ==
|
|
|
|
...
|
|
|
|
== Optimized Logical Plan ==
|
|
|
|
...
|
|
|
|
== Physical Plan ==
|
|
|
|
...
|
|
|
|
>>> sq.stop()
|
|
|
|
"""
|
|
|
|
# Cannot call `_jsq.explain(...)` because it will print in the JVM process.
|
|
|
|
# We should print it in the Python process.
|
|
|
|
print(self._jsq.explainInternal(extended))
|
|
|
|
|
|
|
|
@since(2.1)
|
|
|
|
def exception(self):
|
|
|
|
"""
|
|
|
|
:return: the StreamingQueryException if the query was terminated by an exception, or None.
|
|
|
|
"""
|
|
|
|
if self._jsq.exception().isDefined():
|
|
|
|
je = self._jsq.exception().get()
|
|
|
|
msg = je.toString().split(': ', 1)[1] # Drop the Java StreamingQueryException type info
|
|
|
|
stackTrace = '\n\t at '.join(map(lambda x: x.toString(), je.getStackTrace()))
|
|
|
|
return StreamingQueryException(msg, stackTrace)
|
|
|
|
else:
|
|
|
|
return None
|
|
|
|
|
2016-04-20 13:32:01 -04:00
|
|
|
|
2016-06-15 13:46:02 -04:00
|
|
|
class StreamingQueryManager(object):
|
|
|
|
"""A class to manage all the :class:`StreamingQuery` StreamingQueries active.
|
2016-04-28 18:22:28 -04:00
|
|
|
|
2017-05-26 16:33:23 -04:00
|
|
|
.. note:: Evolving
|
2016-04-28 18:22:28 -04:00
|
|
|
|
|
|
|
.. versionadded:: 2.0
|
|
|
|
"""
|
|
|
|
|
2016-06-15 13:46:02 -04:00
|
|
|
def __init__(self, jsqm):
|
|
|
|
self._jsqm = jsqm
|
2016-04-28 18:22:28 -04:00
|
|
|
|
|
|
|
@property
|
|
|
|
@ignore_unicode_prefix
|
|
|
|
@since(2.0)
|
|
|
|
def active(self):
|
|
|
|
"""Returns a list of active queries associated with this SQLContext
|
|
|
|
|
2016-06-29 01:07:11 -04:00
|
|
|
>>> sq = sdf.writeStream.format('memory').queryName('this_query').start()
|
2016-06-15 13:46:02 -04:00
|
|
|
>>> sqm = spark.streams
|
|
|
|
>>> # get the list of active streaming queries
|
|
|
|
>>> [q.name for q in sqm.active]
|
2016-04-28 18:22:28 -04:00
|
|
|
[u'this_query']
|
2016-06-15 13:46:02 -04:00
|
|
|
>>> sq.stop()
|
2016-04-28 18:22:28 -04:00
|
|
|
"""
|
2016-06-15 13:46:02 -04:00
|
|
|
return [StreamingQuery(jsq) for jsq in self._jsqm.active()]
|
2016-04-28 18:22:28 -04:00
|
|
|
|
2016-06-14 05:12:29 -04:00
|
|
|
@ignore_unicode_prefix
|
2016-04-28 18:22:28 -04:00
|
|
|
@since(2.0)
|
2016-06-14 05:12:29 -04:00
|
|
|
def get(self, id):
|
2016-04-28 18:22:28 -04:00
|
|
|
"""Returns an active query from this SQLContext or throws exception if an active query
|
|
|
|
with this name doesn't exist.
|
|
|
|
|
2016-06-29 01:07:11 -04:00
|
|
|
>>> sq = sdf.writeStream.format('memory').queryName('this_query').start()
|
2016-06-15 13:46:02 -04:00
|
|
|
>>> sq.name
|
2016-06-14 05:12:29 -04:00
|
|
|
u'this_query'
|
2016-06-15 13:46:02 -04:00
|
|
|
>>> sq = spark.streams.get(sq.id)
|
|
|
|
>>> sq.isActive
|
2016-06-14 05:12:29 -04:00
|
|
|
True
|
2016-06-15 13:46:02 -04:00
|
|
|
>>> sq = sqlContext.streams.get(sq.id)
|
|
|
|
>>> sq.isActive
|
2016-04-28 18:22:28 -04:00
|
|
|
True
|
2016-06-15 13:46:02 -04:00
|
|
|
>>> sq.stop()
|
2016-04-28 18:22:28 -04:00
|
|
|
"""
|
2016-06-15 13:46:02 -04:00
|
|
|
return StreamingQuery(self._jsqm.get(id))
|
2016-04-28 18:22:28 -04:00
|
|
|
|
|
|
|
@since(2.0)
|
|
|
|
def awaitAnyTermination(self, timeout=None):
|
|
|
|
"""Wait until any of the queries on the associated SQLContext has terminated since the
|
|
|
|
creation of the context, or since :func:`resetTerminated()` was called. If any query was
|
|
|
|
terminated with an exception, then the exception will be thrown.
|
|
|
|
If `timeout` is set, it returns whether the query has terminated or not within the
|
|
|
|
`timeout` seconds.
|
|
|
|
|
|
|
|
If a query has terminated, then subsequent calls to :func:`awaitAnyTermination()` will
|
|
|
|
either return immediately (if the query was terminated by :func:`query.stop()`),
|
|
|
|
or throw the exception immediately (if the query was terminated with exception). Use
|
|
|
|
:func:`resetTerminated()` to clear past terminations and wait for new terminations.
|
|
|
|
|
|
|
|
In the case where multiple queries have terminated since :func:`resetTermination()`
|
|
|
|
was called, if any query has terminated with exception, then :func:`awaitAnyTermination()`
|
|
|
|
will throw any of the exception. For correctly documenting exceptions across multiple
|
|
|
|
queries, users need to stop all of them after any of them terminates with exception, and
|
|
|
|
then check the `query.exception()` for each query.
|
|
|
|
|
2016-06-15 13:46:02 -04:00
|
|
|
throws :class:`StreamingQueryException`, if `this` query has terminated with an exception
|
2016-04-28 18:22:28 -04:00
|
|
|
"""
|
|
|
|
if timeout is not None:
|
|
|
|
if not isinstance(timeout, (int, float)) or timeout < 0:
|
|
|
|
raise ValueError("timeout must be a positive integer or float. Got %s" % timeout)
|
2016-06-15 13:46:02 -04:00
|
|
|
return self._jsqm.awaitAnyTermination(int(timeout * 1000))
|
2016-04-28 18:22:28 -04:00
|
|
|
else:
|
2016-06-15 13:46:02 -04:00
|
|
|
return self._jsqm.awaitAnyTermination()
|
2016-04-28 18:22:28 -04:00
|
|
|
|
|
|
|
@since(2.0)
|
|
|
|
def resetTerminated(self):
|
|
|
|
"""Forget about past terminated queries so that :func:`awaitAnyTermination()` can be used
|
|
|
|
again to wait for new terminations.
|
|
|
|
|
2016-06-14 05:12:29 -04:00
|
|
|
>>> spark.streams.resetTerminated()
|
2016-04-28 18:22:28 -04:00
|
|
|
"""
|
2016-06-15 13:46:02 -04:00
|
|
|
self._jsqm.resetTerminated()
|
2016-04-28 18:22:28 -04:00
|
|
|
|
|
|
|
|
2016-06-29 01:07:11 -04:00
|
|
|
class DataStreamReader(OptionUtils):
|
|
|
|
"""
|
|
|
|
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.
|
|
|
|
|
2017-05-26 16:33:23 -04:00
|
|
|
.. note:: Evolving.
|
2016-06-29 01:07:11 -04:00
|
|
|
|
|
|
|
.. 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.
|
|
|
|
|
2017-05-26 16:33:23 -04:00
|
|
|
.. note:: Evolving.
|
2016-06-29 01:07:11 -04:00
|
|
|
|
|
|
|
:param source: string, name of the data source, e.g. 'json', 'parquet'.
|
|
|
|
|
|
|
|
>>> s = spark.readStream.format("text")
|
|
|
|
"""
|
|
|
|
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.
|
|
|
|
|
2017-05-26 16:33:23 -04:00
|
|
|
.. note:: Evolving.
|
2016-06-29 01:07:11 -04:00
|
|
|
|
2017-06-23 23:39:41 -04:00
|
|
|
:param schema: a :class:`pyspark.sql.types.StructType` object or a DDL-formatted string
|
|
|
|
(For example ``col0 INT, col1 DOUBLE``).
|
2016-06-29 01:07:11 -04:00
|
|
|
|
|
|
|
>>> s = spark.readStream.schema(sdf_schema)
|
2017-06-23 23:39:41 -04:00
|
|
|
>>> s = spark.readStream.schema("col0 INT, col1 DOUBLE")
|
2016-06-29 01:07:11 -04:00
|
|
|
"""
|
2016-08-25 02:36:04 -04:00
|
|
|
from pyspark.sql import SparkSession
|
|
|
|
spark = SparkSession.builder.getOrCreate()
|
2017-06-23 23:39:41 -04:00
|
|
|
if isinstance(schema, StructType):
|
|
|
|
jschema = spark._jsparkSession.parseDataType(schema.json())
|
|
|
|
self._jreader = self._jreader.schema(jschema)
|
|
|
|
elif isinstance(schema, basestring):
|
|
|
|
self._jreader = self._jreader.schema(schema)
|
|
|
|
else:
|
|
|
|
raise TypeError("schema should be StructType or string")
|
2016-06-29 01:07:11 -04:00
|
|
|
return self
|
|
|
|
|
|
|
|
@since(2.0)
|
|
|
|
def option(self, key, value):
|
|
|
|
"""Adds an input option for the underlying data source.
|
|
|
|
|
2017-03-15 01:30:16 -04:00
|
|
|
You can set the following option(s) for reading files:
|
|
|
|
* ``timeZone``: sets the string that indicates a timezone to be used to parse timestamps
|
|
|
|
in the JSON/CSV datasources or partition values.
|
|
|
|
If it isn't set, it uses the default value, session local timezone.
|
|
|
|
|
2017-05-26 16:33:23 -04:00
|
|
|
.. note:: Evolving.
|
2016-06-29 01:07:11 -04:00
|
|
|
|
|
|
|
>>> s = spark.readStream.option("x", 1)
|
|
|
|
"""
|
|
|
|
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.
|
|
|
|
|
2017-03-15 01:30:16 -04:00
|
|
|
You can set the following option(s) for reading files:
|
|
|
|
* ``timeZone``: sets the string that indicates a timezone to be used to parse timestamps
|
|
|
|
in the JSON/CSV datasources or partition values.
|
|
|
|
If it isn't set, it uses the default value, session local timezone.
|
|
|
|
|
2017-05-26 16:33:23 -04:00
|
|
|
.. note:: Evolving.
|
2016-06-29 01:07:11 -04:00
|
|
|
|
|
|
|
>>> s = spark.readStream.options(x="1", y=2)
|
|
|
|
"""
|
|
|
|
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`.
|
|
|
|
|
2017-05-26 16:33:23 -04:00
|
|
|
.. note:: Evolving.
|
2016-06-29 01:07:11 -04:00
|
|
|
|
|
|
|
:param path: optional string for file-system backed data sources.
|
|
|
|
:param format: optional string for format of the data source. Default to 'parquet'.
|
2017-06-23 23:39:41 -04:00
|
|
|
:param schema: optional :class:`pyspark.sql.types.StructType` for the input schema
|
|
|
|
or a DDL-formatted string (For example ``col0 INT, col1 DOUBLE``).
|
2016-06-29 01:07:11 -04:00
|
|
|
:param options: all other string options
|
|
|
|
|
[MINOR][PYSPARK][DOCS] Fix examples in PySpark documentation
## What changes were proposed in this pull request?
This PR proposes to fix wrongly indented examples in PySpark documentation
```
- >>> json_sdf = spark.readStream.format("json")\
- .schema(sdf_schema)\
- .load(tempfile.mkdtemp())
+ >>> json_sdf = spark.readStream.format("json") \\
+ ... .schema(sdf_schema) \\
+ ... .load(tempfile.mkdtemp())
```
```
- people.filter(people.age > 30).join(department, people.deptId == department.id)\
+ people.filter(people.age > 30).join(department, people.deptId == department.id) \\
```
```
- >>> examples = [LabeledPoint(1.1, Vectors.sparse(3, [(0, 1.23), (2, 4.56)])), \
- LabeledPoint(0.0, Vectors.dense([1.01, 2.02, 3.03]))]
+ >>> examples = [LabeledPoint(1.1, Vectors.sparse(3, [(0, 1.23), (2, 4.56)])),
+ ... LabeledPoint(0.0, Vectors.dense([1.01, 2.02, 3.03]))]
```
```
- >>> examples = [LabeledPoint(1.1, Vectors.sparse(3, [(0, -1.23), (2, 4.56e-7)])), \
- LabeledPoint(0.0, Vectors.dense([1.01, 2.02, 3.03]))]
+ >>> examples = [LabeledPoint(1.1, Vectors.sparse(3, [(0, -1.23), (2, 4.56e-7)])),
+ ... LabeledPoint(0.0, Vectors.dense([1.01, 2.02, 3.03]))]
```
```
- ... for x in iterator:
- ... print(x)
+ ... for x in iterator:
+ ... print(x)
```
## How was this patch tested?
Manually tested.
**Before**
![2016-09-26 8 36 02](https://cloud.githubusercontent.com/assets/6477701/18834471/05c7a478-8431-11e6-94bb-09aa37b12ddb.png)
![2016-09-26 9 22 16](https://cloud.githubusercontent.com/assets/6477701/18834472/06c8735c-8431-11e6-8775-78631eab0411.png)
<img width="601" alt="2016-09-27 2 29 27" src="https://cloud.githubusercontent.com/assets/6477701/18861294/29c0d5b4-84bf-11e6-99c5-3c9d913c125d.png">
<img width="1056" alt="2016-09-27 2 29 58" src="https://cloud.githubusercontent.com/assets/6477701/18861298/31694cd8-84bf-11e6-9e61-9888cb8c2089.png">
<img width="1079" alt="2016-09-27 2 30 05" src="https://cloud.githubusercontent.com/assets/6477701/18861301/359722da-84bf-11e6-97f9-5f5365582d14.png">
**After**
![2016-09-26 9 29 47](https://cloud.githubusercontent.com/assets/6477701/18834467/0367f9da-8431-11e6-86d9-a490d3297339.png)
![2016-09-26 9 30 24](https://cloud.githubusercontent.com/assets/6477701/18834463/f870fae0-8430-11e6-9482-01fc47898492.png)
<img width="515" alt="2016-09-27 2 28 19" src="https://cloud.githubusercontent.com/assets/6477701/18861305/3ff88b88-84bf-11e6-902c-9f725e8a8b10.png">
<img width="652" alt="2016-09-27 3 50 59" src="https://cloud.githubusercontent.com/assets/6477701/18863053/592fbc74-84ca-11e6-8dbf-99cf57947de8.png">
<img width="709" alt="2016-09-27 3 51 03" src="https://cloud.githubusercontent.com/assets/6477701/18863060/601607be-84ca-11e6-80aa-a401df41c321.png">
Author: hyukjinkwon <gurwls223@gmail.com>
Closes #15242 from HyukjinKwon/minor-example-pyspark.
2016-09-28 06:19:04 -04:00
|
|
|
>>> json_sdf = spark.readStream.format("json") \\
|
|
|
|
... .schema(sdf_schema) \\
|
|
|
|
... .load(tempfile.mkdtemp())
|
2016-06-29 01:07:11 -04:00
|
|
|
>>> json_sdf.isStreaming
|
|
|
|
True
|
|
|
|
>>> json_sdf.schema == sdf_schema
|
|
|
|
True
|
|
|
|
"""
|
|
|
|
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,
|
2017-02-16 23:51:19 -05:00
|
|
|
mode=None, columnNameOfCorruptRecord=None, dateFormat=None, timestampFormat=None,
|
2018-03-28 07:49:27 -04:00
|
|
|
multiLine=None, allowUnquotedControlChars=None, lineSep=None):
|
2016-06-29 01:07:11 -04:00
|
|
|
"""
|
2017-02-16 23:51:19 -05:00
|
|
|
Loads a JSON file stream and returns the results as a :class:`DataFrame`.
|
|
|
|
|
2017-04-12 04:16:39 -04:00
|
|
|
`JSON Lines <http://jsonlines.org/>`_ (newline-delimited JSON) is supported by default.
|
2017-06-15 01:18:19 -04:00
|
|
|
For JSON (one record per file), set the ``multiLine`` parameter to ``true``.
|
2016-06-29 01:07:11 -04:00
|
|
|
|
|
|
|
If the ``schema`` parameter is not specified, this function goes
|
|
|
|
through the input once to determine the input schema.
|
|
|
|
|
2017-05-26 16:33:23 -04:00
|
|
|
.. note:: Evolving.
|
2016-06-29 01:07:11 -04:00
|
|
|
|
|
|
|
:param path: string represents path to the JSON dataset,
|
|
|
|
or RDD of Strings storing JSON objects.
|
2017-06-23 23:39:41 -04:00
|
|
|
:param schema: an optional :class:`pyspark.sql.types.StructType` for the input schema
|
|
|
|
or a DDL-formatted string (For example ``col0 INT, col1 DOUBLE``).
|
2016-06-29 01:07:11 -04:00
|
|
|
: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``.
|
|
|
|
|
2018-02-27 21:00:54 -05:00
|
|
|
* ``PERMISSIVE`` : when it meets a corrupted record, puts the malformed string \
|
|
|
|
into a field configured by ``columnNameOfCorruptRecord``, and sets other \
|
|
|
|
fields to ``null``. To keep corrupt records, an user can set a string type \
|
|
|
|
field named ``columnNameOfCorruptRecord`` in an user-defined schema. If a \
|
|
|
|
schema does not have the field, it drops corrupt records during parsing. \
|
|
|
|
When inferring a schema, it implicitly adds a ``columnNameOfCorruptRecord`` \
|
|
|
|
field in an output schema.
|
2016-06-29 01:07:11 -04:00
|
|
|
* ``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``.
|
2016-08-24 16:16:20 -04:00
|
|
|
:param dateFormat: sets the string that indicates a date format. Custom date formats
|
|
|
|
follow the formats at ``java.text.SimpleDateFormat``. This
|
|
|
|
applies to date type. If None is set, it uses the
|
[SPARK-18937][SQL] Timezone support in CSV/JSON parsing
## What changes were proposed in this pull request?
This is a follow-up pr of #16308.
This pr enables timezone support in CSV/JSON parsing.
We should introduce `timeZone` option for CSV/JSON datasources (the default value of the option is session local timezone).
The datasources should use the `timeZone` option to format/parse to write/read timestamp values.
Notice that while reading, if the timestampFormat has the timezone info, the timezone will not be used because we should respect the timezone in the values.
For example, if you have timestamp `"2016-01-01 00:00:00"` in `GMT`, the values written with the default timezone option, which is `"GMT"` because session local timezone is `"GMT"` here, are:
```scala
scala> spark.conf.set("spark.sql.session.timeZone", "GMT")
scala> val df = Seq(new java.sql.Timestamp(1451606400000L)).toDF("ts")
df: org.apache.spark.sql.DataFrame = [ts: timestamp]
scala> df.show()
+-------------------+
|ts |
+-------------------+
|2016-01-01 00:00:00|
+-------------------+
scala> df.write.json("/path/to/gmtjson")
```
```sh
$ cat /path/to/gmtjson/part-*
{"ts":"2016-01-01T00:00:00.000Z"}
```
whereas setting the option to `"PST"`, they are:
```scala
scala> df.write.option("timeZone", "PST").json("/path/to/pstjson")
```
```sh
$ cat /path/to/pstjson/part-*
{"ts":"2015-12-31T16:00:00.000-08:00"}
```
We can properly read these files even if the timezone option is wrong because the timestamp values have timezone info:
```scala
scala> val schema = new StructType().add("ts", TimestampType)
schema: org.apache.spark.sql.types.StructType = StructType(StructField(ts,TimestampType,true))
scala> spark.read.schema(schema).json("/path/to/gmtjson").show()
+-------------------+
|ts |
+-------------------+
|2016-01-01 00:00:00|
+-------------------+
scala> spark.read.schema(schema).option("timeZone", "PST").json("/path/to/gmtjson").show()
+-------------------+
|ts |
+-------------------+
|2016-01-01 00:00:00|
+-------------------+
```
And even if `timezoneFormat` doesn't contain timezone info, we can properly read the values with setting correct timezone option:
```scala
scala> df.write.option("timestampFormat", "yyyy-MM-dd'T'HH:mm:ss").option("timeZone", "JST").json("/path/to/jstjson")
```
```sh
$ cat /path/to/jstjson/part-*
{"ts":"2016-01-01T09:00:00"}
```
```scala
// wrong result
scala> spark.read.schema(schema).option("timestampFormat", "yyyy-MM-dd'T'HH:mm:ss").json("/path/to/jstjson").show()
+-------------------+
|ts |
+-------------------+
|2016-01-01 09:00:00|
+-------------------+
// correct result
scala> spark.read.schema(schema).option("timestampFormat", "yyyy-MM-dd'T'HH:mm:ss").option("timeZone", "JST").json("/path/to/jstjson").show()
+-------------------+
|ts |
+-------------------+
|2016-01-01 00:00:00|
+-------------------+
```
This pr also makes `JsonToStruct` and `StructToJson` `TimeZoneAwareExpression` to be able to evaluate values with timezone option.
## How was this patch tested?
Existing tests and added some tests.
Author: Takuya UESHIN <ueshin@happy-camper.st>
Closes #16750 from ueshin/issues/SPARK-18937.
2017-02-15 16:26:34 -05:00
|
|
|
default value, ``yyyy-MM-dd``.
|
2016-08-24 16:16:20 -04:00
|
|
|
:param timestampFormat: sets the string that indicates a timestamp format. Custom date
|
|
|
|
formats follow the formats at ``java.text.SimpleDateFormat``.
|
|
|
|
This applies to timestamp type. If None is set, it uses the
|
2017-04-03 05:07:41 -04:00
|
|
|
default value, ``yyyy-MM-dd'T'HH:mm:ss.SSSXXX``.
|
2017-06-15 01:18:19 -04:00
|
|
|
:param multiLine: parse one record, which may span multiple lines, per file. If None is
|
2017-02-16 23:51:19 -05:00
|
|
|
set, it uses the default value, ``false``.
|
2017-08-25 13:18:03 -04:00
|
|
|
:param allowUnquotedControlChars: allows JSON Strings to contain unquoted control
|
|
|
|
characters (ASCII characters with value less than 32,
|
|
|
|
including tab and line feed characters) or not.
|
2018-03-28 07:49:27 -04:00
|
|
|
:param lineSep: defines the line separator that should be used for parsing. If None is
|
|
|
|
set, it covers all ``\\r``, ``\\r\\n`` and ``\\n``.
|
2016-06-29 01:07:11 -04:00
|
|
|
|
2016-07-01 18:16:04 -04:00
|
|
|
>>> json_sdf = spark.readStream.json(tempfile.mkdtemp(), schema = sdf_schema)
|
2016-06-29 01:07:11 -04:00
|
|
|
>>> json_sdf.isStreaming
|
|
|
|
True
|
|
|
|
>>> json_sdf.schema == sdf_schema
|
|
|
|
True
|
|
|
|
"""
|
|
|
|
self._set_opts(
|
|
|
|
schema=schema, primitivesAsString=primitivesAsString, prefersDecimal=prefersDecimal,
|
|
|
|
allowComments=allowComments, allowUnquotedFieldNames=allowUnquotedFieldNames,
|
|
|
|
allowSingleQuotes=allowSingleQuotes, allowNumericLeadingZero=allowNumericLeadingZero,
|
|
|
|
allowBackslashEscapingAnyCharacter=allowBackslashEscapingAnyCharacter,
|
2016-08-24 16:16:20 -04:00
|
|
|
mode=mode, columnNameOfCorruptRecord=columnNameOfCorruptRecord, dateFormat=dateFormat,
|
2017-08-25 13:18:03 -04:00
|
|
|
timestampFormat=timestampFormat, multiLine=multiLine,
|
2018-03-28 07:49:27 -04:00
|
|
|
allowUnquotedControlChars=allowUnquotedControlChars, lineSep=lineSep)
|
2016-06-29 01:07:11 -04:00
|
|
|
if isinstance(path, basestring):
|
|
|
|
return self._df(self._jreader.json(path))
|
|
|
|
else:
|
|
|
|
raise TypeError("path can be only a single string")
|
|
|
|
|
2017-12-20 02:50:06 -05:00
|
|
|
@since(2.3)
|
|
|
|
def orc(self, path):
|
|
|
|
"""Loads a ORC file stream, returning the result as a :class:`DataFrame`.
|
|
|
|
|
|
|
|
.. note:: Evolving.
|
|
|
|
|
|
|
|
>>> orc_sdf = spark.readStream.schema(sdf_schema).orc(tempfile.mkdtemp())
|
|
|
|
>>> orc_sdf.isStreaming
|
|
|
|
True
|
|
|
|
>>> orc_sdf.schema == sdf_schema
|
|
|
|
True
|
|
|
|
"""
|
|
|
|
if isinstance(path, basestring):
|
|
|
|
return self._df(self._jreader.orc(path))
|
|
|
|
else:
|
|
|
|
raise TypeError("path can be only a single string")
|
|
|
|
|
2016-06-29 01:07:11 -04:00
|
|
|
@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``.
|
|
|
|
|
2017-05-26 16:33:23 -04:00
|
|
|
.. note:: Evolving.
|
2016-06-29 01:07:11 -04:00
|
|
|
|
2016-07-01 18:16:04 -04:00
|
|
|
>>> parquet_sdf = spark.readStream.schema(sdf_schema).parquet(tempfile.mkdtemp())
|
2016-06-29 01:07:11 -04:00
|
|
|
>>> parquet_sdf.isStreaming
|
|
|
|
True
|
|
|
|
>>> parquet_sdf.schema == sdf_schema
|
|
|
|
True
|
|
|
|
"""
|
|
|
|
if isinstance(path, basestring):
|
|
|
|
return self._df(self._jreader.parquet(path))
|
|
|
|
else:
|
|
|
|
raise TypeError("path can be only a single string")
|
|
|
|
|
|
|
|
@ignore_unicode_prefix
|
|
|
|
@since(2.0)
|
2018-03-21 12:46:47 -04:00
|
|
|
def text(self, path, wholetext=False, lineSep=None):
|
2016-06-29 01:07:11 -04:00
|
|
|
"""
|
|
|
|
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.
|
|
|
|
|
2018-03-21 12:46:47 -04:00
|
|
|
By default, each line in the text file is a new row in the resulting DataFrame.
|
2016-06-29 01:07:11 -04:00
|
|
|
|
2017-05-26 16:33:23 -04:00
|
|
|
.. note:: Evolving.
|
2016-06-29 01:07:11 -04:00
|
|
|
|
|
|
|
:param paths: string, or list of strings, for input path(s).
|
2018-03-21 12:46:47 -04:00
|
|
|
:param wholetext: if true, read each file from input path(s) as a single row.
|
|
|
|
:param lineSep: defines the line separator that should be used for parsing. If None is
|
|
|
|
set, it covers all ``\\r``, ``\\r\\n`` and ``\\n``.
|
2016-06-29 01:07:11 -04:00
|
|
|
|
2016-06-30 19:51:11 -04:00
|
|
|
>>> text_sdf = spark.readStream.text(tempfile.mkdtemp())
|
2016-06-29 01:07:11 -04:00
|
|
|
>>> text_sdf.isStreaming
|
|
|
|
True
|
|
|
|
>>> "value" in str(text_sdf.schema)
|
|
|
|
True
|
|
|
|
"""
|
2018-03-21 12:46:47 -04:00
|
|
|
self._set_opts(wholetext=wholetext, lineSep=lineSep)
|
2016-06-29 01:07:11 -04:00
|
|
|
if isinstance(path, basestring):
|
|
|
|
return self._df(self._jreader.text(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,
|
|
|
|
comment=None, header=None, inferSchema=None, ignoreLeadingWhiteSpace=None,
|
|
|
|
ignoreTrailingWhiteSpace=None, nullValue=None, nanValue=None, positiveInf=None,
|
2016-08-24 16:16:20 -04:00
|
|
|
negativeInf=None, dateFormat=None, timestampFormat=None, maxColumns=None,
|
2017-03-15 01:30:16 -04:00
|
|
|
maxCharsPerColumn=None, maxMalformedLogPerPartition=None, mode=None,
|
2017-12-28 18:30:06 -05:00
|
|
|
columnNameOfCorruptRecord=None, multiLine=None, charToEscapeQuoteEscaping=None):
|
2016-06-29 01:07:11 -04:00
|
|
|
"""Loads a CSV file stream and returns the result as a :class:`DataFrame`.
|
|
|
|
|
|
|
|
This function will go through the input once to determine the input schema if
|
|
|
|
``inferSchema`` is enabled. To avoid going through the entire data once, disable
|
|
|
|
``inferSchema`` option or specify the schema explicitly using ``schema``.
|
|
|
|
|
2017-05-26 16:33:23 -04:00
|
|
|
.. note:: Evolving.
|
2016-06-29 01:07:11 -04:00
|
|
|
|
|
|
|
:param path: string, or list of strings, for input path(s).
|
2017-06-23 23:39:41 -04:00
|
|
|
:param schema: an optional :class:`pyspark.sql.types.StructType` for the input schema
|
|
|
|
or a DDL-formatted string (For example ``col0 INT, col1 DOUBLE``).
|
2017-12-28 18:30:06 -05:00
|
|
|
:param sep: sets a single character as a separator for each field and value.
|
2016-06-29 01:07:11 -04:00
|
|
|
If None is set, it uses the default value, ``,``.
|
|
|
|
:param encoding: decodes the CSV files by the given encoding type. If None is set,
|
|
|
|
it uses the default value, ``UTF-8``.
|
2017-12-28 18:30:06 -05:00
|
|
|
:param quote: sets a single character used for escaping quoted values where the
|
2016-06-29 01:07:11 -04:00
|
|
|
separator can be part of the value. If None is set, it uses the default
|
|
|
|
value, ``"``. If you would like to turn off quotations, you need to set an
|
|
|
|
empty string.
|
2017-12-28 18:30:06 -05:00
|
|
|
:param escape: sets a single character used for escaping quotes inside an already
|
2016-06-29 01:07:11 -04:00
|
|
|
quoted value. If None is set, it uses the default value, ``\``.
|
2017-12-28 18:30:06 -05:00
|
|
|
:param comment: sets a single character used for skipping lines beginning with this
|
2016-06-29 01:07:11 -04:00
|
|
|
character. By default (None), it is disabled.
|
|
|
|
:param header: uses the first line as names of columns. If None is set, it uses the
|
|
|
|
default value, ``false``.
|
|
|
|
:param inferSchema: infers the input schema automatically from data. It requires one extra
|
|
|
|
pass over the data. If None is set, it uses the default value, ``false``.
|
[SPARK-18579][SQL] Use ignoreLeadingWhiteSpace and ignoreTrailingWhiteSpace options in CSV writing
## What changes were proposed in this pull request?
This PR proposes to support _not_ trimming the white spaces when writing out. These are `false` by default in CSV reading path but these are `true` by default in CSV writing in univocity parser.
Both `ignoreLeadingWhiteSpace` and `ignoreTrailingWhiteSpace` options are not being used for writing and therefore, we are always trimming the white spaces.
It seems we should provide a way to keep this white spaces easily.
WIth the data below:
```scala
val df = spark.read.csv(Seq("a , b , c").toDS)
df.show()
```
```
+---+----+---+
|_c0| _c1|_c2|
+---+----+---+
| a | b | c|
+---+----+---+
```
**Before**
```scala
df.write.csv("/tmp/text.csv")
spark.read.text("/tmp/text.csv").show()
```
```
+-----+
|value|
+-----+
|a,b,c|
+-----+
```
It seems this can't be worked around via `quoteAll` too.
```scala
df.write.option("quoteAll", true).csv("/tmp/text.csv")
spark.read.text("/tmp/text.csv").show()
```
```
+-----------+
| value|
+-----------+
|"a","b","c"|
+-----------+
```
**After**
```scala
df.write.option("ignoreLeadingWhiteSpace", false).option("ignoreTrailingWhiteSpace", false).csv("/tmp/text.csv")
spark.read.text("/tmp/text.csv").show()
```
```
+----------+
| value|
+----------+
|a , b , c|
+----------+
```
Note that this case is possible in R
```r
> system("cat text.csv")
f1,f2,f3
a , b , c
> df <- read.csv(file="text.csv")
> df
f1 f2 f3
1 a b c
> write.csv(df, file="text1.csv", quote=F, row.names=F)
> system("cat text1.csv")
f1,f2,f3
a , b , c
```
## How was this patch tested?
Unit tests in `CSVSuite` and manual tests for Python.
Author: hyukjinkwon <gurwls223@gmail.com>
Closes #17310 from HyukjinKwon/SPARK-18579.
2017-03-23 03:25:01 -04:00
|
|
|
:param ignoreLeadingWhiteSpace: a flag indicating whether or not leading whitespaces from
|
|
|
|
values being read should be skipped. If None is set, it
|
|
|
|
uses the default value, ``false``.
|
|
|
|
:param ignoreTrailingWhiteSpace: a flag indicating whether or not trailing whitespaces from
|
|
|
|
values being read should be skipped. If None is set, it
|
|
|
|
uses the default value, ``false``.
|
2016-06-29 01:07:11 -04:00
|
|
|
:param nullValue: sets the string representation of a null value. If None is set, it uses
|
2016-09-18 14:25:58 -04:00
|
|
|
the default value, empty string. Since 2.0.1, this ``nullValue`` param
|
|
|
|
applies to all supported types including the string type.
|
2016-06-29 01:07:11 -04:00
|
|
|
:param nanValue: sets the string representation of a non-number value. If None is set, it
|
|
|
|
uses the default value, ``NaN``.
|
|
|
|
:param positiveInf: sets the string representation of a positive infinity value. If None
|
|
|
|
is set, it uses the default value, ``Inf``.
|
|
|
|
:param negativeInf: sets the string representation of a negative infinity value. If None
|
|
|
|
is set, it uses the default value, ``Inf``.
|
|
|
|
:param dateFormat: sets the string that indicates a date format. Custom date formats
|
|
|
|
follow the formats at ``java.text.SimpleDateFormat``. This
|
2016-08-24 16:16:20 -04:00
|
|
|
applies to date type. If None is set, it uses the
|
[SPARK-18937][SQL] Timezone support in CSV/JSON parsing
## What changes were proposed in this pull request?
This is a follow-up pr of #16308.
This pr enables timezone support in CSV/JSON parsing.
We should introduce `timeZone` option for CSV/JSON datasources (the default value of the option is session local timezone).
The datasources should use the `timeZone` option to format/parse to write/read timestamp values.
Notice that while reading, if the timestampFormat has the timezone info, the timezone will not be used because we should respect the timezone in the values.
For example, if you have timestamp `"2016-01-01 00:00:00"` in `GMT`, the values written with the default timezone option, which is `"GMT"` because session local timezone is `"GMT"` here, are:
```scala
scala> spark.conf.set("spark.sql.session.timeZone", "GMT")
scala> val df = Seq(new java.sql.Timestamp(1451606400000L)).toDF("ts")
df: org.apache.spark.sql.DataFrame = [ts: timestamp]
scala> df.show()
+-------------------+
|ts |
+-------------------+
|2016-01-01 00:00:00|
+-------------------+
scala> df.write.json("/path/to/gmtjson")
```
```sh
$ cat /path/to/gmtjson/part-*
{"ts":"2016-01-01T00:00:00.000Z"}
```
whereas setting the option to `"PST"`, they are:
```scala
scala> df.write.option("timeZone", "PST").json("/path/to/pstjson")
```
```sh
$ cat /path/to/pstjson/part-*
{"ts":"2015-12-31T16:00:00.000-08:00"}
```
We can properly read these files even if the timezone option is wrong because the timestamp values have timezone info:
```scala
scala> val schema = new StructType().add("ts", TimestampType)
schema: org.apache.spark.sql.types.StructType = StructType(StructField(ts,TimestampType,true))
scala> spark.read.schema(schema).json("/path/to/gmtjson").show()
+-------------------+
|ts |
+-------------------+
|2016-01-01 00:00:00|
+-------------------+
scala> spark.read.schema(schema).option("timeZone", "PST").json("/path/to/gmtjson").show()
+-------------------+
|ts |
+-------------------+
|2016-01-01 00:00:00|
+-------------------+
```
And even if `timezoneFormat` doesn't contain timezone info, we can properly read the values with setting correct timezone option:
```scala
scala> df.write.option("timestampFormat", "yyyy-MM-dd'T'HH:mm:ss").option("timeZone", "JST").json("/path/to/jstjson")
```
```sh
$ cat /path/to/jstjson/part-*
{"ts":"2016-01-01T09:00:00"}
```
```scala
// wrong result
scala> spark.read.schema(schema).option("timestampFormat", "yyyy-MM-dd'T'HH:mm:ss").json("/path/to/jstjson").show()
+-------------------+
|ts |
+-------------------+
|2016-01-01 09:00:00|
+-------------------+
// correct result
scala> spark.read.schema(schema).option("timestampFormat", "yyyy-MM-dd'T'HH:mm:ss").option("timeZone", "JST").json("/path/to/jstjson").show()
+-------------------+
|ts |
+-------------------+
|2016-01-01 00:00:00|
+-------------------+
```
This pr also makes `JsonToStruct` and `StructToJson` `TimeZoneAwareExpression` to be able to evaluate values with timezone option.
## How was this patch tested?
Existing tests and added some tests.
Author: Takuya UESHIN <ueshin@happy-camper.st>
Closes #16750 from ueshin/issues/SPARK-18937.
2017-02-15 16:26:34 -05:00
|
|
|
default value, ``yyyy-MM-dd``.
|
2016-08-24 16:16:20 -04:00
|
|
|
:param timestampFormat: sets the string that indicates a timestamp format. Custom date
|
|
|
|
formats follow the formats at ``java.text.SimpleDateFormat``.
|
|
|
|
This applies to timestamp type. If None is set, it uses the
|
2017-04-03 05:07:41 -04:00
|
|
|
default value, ``yyyy-MM-dd'T'HH:mm:ss.SSSXXX``.
|
2016-06-29 01:07:11 -04:00
|
|
|
:param maxColumns: defines a hard limit of how many columns a record can have. If None is
|
|
|
|
set, it uses the default value, ``20480``.
|
|
|
|
:param maxCharsPerColumn: defines the maximum number of characters allowed for any given
|
|
|
|
value being read. If None is set, it uses the default value,
|
2016-09-21 05:35:29 -04:00
|
|
|
``-1`` meaning unlimited length.
|
2017-03-22 12:52:37 -04:00
|
|
|
:param maxMalformedLogPerPartition: this parameter is no longer used since Spark 2.2.0.
|
|
|
|
If specified, it is ignored.
|
2016-06-29 01:07:11 -04:00
|
|
|
:param mode: allows a mode for dealing with corrupt records during parsing. If None is
|
|
|
|
set, it uses the default value, ``PERMISSIVE``.
|
|
|
|
|
2018-02-27 21:00:54 -05:00
|
|
|
* ``PERMISSIVE`` : when it meets a corrupted record, puts the malformed string \
|
|
|
|
into a field configured by ``columnNameOfCorruptRecord``, and sets other \
|
|
|
|
fields to ``null``. To keep corrupt records, an user can set a string type \
|
|
|
|
field named ``columnNameOfCorruptRecord`` in an user-defined schema. If a \
|
|
|
|
schema does not have the field, it drops corrupt records during parsing. \
|
|
|
|
A record with less/more tokens than schema is not a corrupted record to CSV. \
|
|
|
|
When it meets a record having fewer tokens than the length of the schema, \
|
|
|
|
sets ``null`` to extra fields. When the record has more tokens than the \
|
|
|
|
length of the schema, it drops extra tokens.
|
2016-06-29 01:07:11 -04:00
|
|
|
* ``DROPMALFORMED`` : ignores the whole corrupted records.
|
|
|
|
* ``FAILFAST`` : throws an exception when it meets corrupted records.
|
|
|
|
|
2017-02-23 15:09:36 -05:00
|
|
|
: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``.
|
2017-06-15 01:18:19 -04:00
|
|
|
:param multiLine: parse one record, which may span multiple lines. If None is
|
2017-02-28 16:34:33 -05:00
|
|
|
set, it uses the default value, ``false``.
|
2017-12-28 18:30:06 -05:00
|
|
|
:param charToEscapeQuoteEscaping: sets a single character used for escaping the escape for
|
|
|
|
the quote character. If None is set, the default value is
|
|
|
|
escape character when escape and quote characters are
|
|
|
|
different, ``\0`` otherwise..
|
2017-02-23 15:09:36 -05:00
|
|
|
|
2016-07-01 18:16:04 -04:00
|
|
|
>>> csv_sdf = spark.readStream.csv(tempfile.mkdtemp(), schema = sdf_schema)
|
2016-06-29 01:07:11 -04:00
|
|
|
>>> csv_sdf.isStreaming
|
|
|
|
True
|
|
|
|
>>> csv_sdf.schema == sdf_schema
|
|
|
|
True
|
|
|
|
"""
|
|
|
|
self._set_opts(
|
|
|
|
schema=schema, sep=sep, encoding=encoding, quote=quote, escape=escape, comment=comment,
|
|
|
|
header=header, inferSchema=inferSchema, ignoreLeadingWhiteSpace=ignoreLeadingWhiteSpace,
|
|
|
|
ignoreTrailingWhiteSpace=ignoreTrailingWhiteSpace, nullValue=nullValue,
|
|
|
|
nanValue=nanValue, positiveInf=positiveInf, negativeInf=negativeInf,
|
2016-08-24 16:16:20 -04:00
|
|
|
dateFormat=dateFormat, timestampFormat=timestampFormat, maxColumns=maxColumns,
|
|
|
|
maxCharsPerColumn=maxCharsPerColumn,
|
2017-03-15 01:30:16 -04:00
|
|
|
maxMalformedLogPerPartition=maxMalformedLogPerPartition, mode=mode,
|
2017-12-28 18:30:06 -05:00
|
|
|
columnNameOfCorruptRecord=columnNameOfCorruptRecord, multiLine=multiLine,
|
|
|
|
charToEscapeQuoteEscaping=charToEscapeQuoteEscaping)
|
2016-06-29 01:07:11 -04:00
|
|
|
if isinstance(path, basestring):
|
|
|
|
return self._df(self._jreader.csv(path))
|
|
|
|
else:
|
|
|
|
raise TypeError("path can be only a single string")
|
|
|
|
|
|
|
|
|
|
|
|
class DataStreamWriter(object):
|
|
|
|
"""
|
|
|
|
Interface used to write a streaming :class:`DataFrame` to external storage systems
|
|
|
|
(e.g. file systems, key-value stores, etc). Use :func:`DataFrame.writeStream`
|
|
|
|
to access this.
|
|
|
|
|
2017-05-26 16:33:23 -04:00
|
|
|
.. note:: Evolving.
|
2016-06-29 01:07:11 -04:00
|
|
|
|
|
|
|
.. versionadded:: 2.0
|
|
|
|
"""
|
|
|
|
|
|
|
|
def __init__(self, df):
|
|
|
|
self._df = df
|
|
|
|
self._spark = df.sql_ctx
|
|
|
|
self._jwrite = df._jdf.writeStream()
|
|
|
|
|
|
|
|
def _sq(self, jsq):
|
|
|
|
from pyspark.sql.streaming import StreamingQuery
|
|
|
|
return StreamingQuery(jsq)
|
|
|
|
|
|
|
|
@since(2.0)
|
|
|
|
def outputMode(self, outputMode):
|
|
|
|
"""Specifies how data of a streaming DataFrame/Dataset is written to a streaming sink.
|
|
|
|
|
|
|
|
Options include:
|
|
|
|
|
|
|
|
* `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
|
2017-01-10 20:58:11 -05:00
|
|
|
* `update`:only the rows that were updated in the streaming DataFrame/Dataset will be
|
|
|
|
written to the sink every time there are some updates. If the query doesn't contain
|
|
|
|
aggregations, it will be equivalent to `append` mode.
|
2016-06-29 01:07:11 -04:00
|
|
|
|
2017-05-26 16:33:23 -04:00
|
|
|
.. note:: Evolving.
|
2016-06-29 01:07:11 -04:00
|
|
|
|
|
|
|
>>> writer = sdf.writeStream.outputMode('append')
|
|
|
|
"""
|
|
|
|
if not outputMode or type(outputMode) != str or len(outputMode.strip()) == 0:
|
|
|
|
raise ValueError('The output mode must be a non-empty string. Got: %s' % outputMode)
|
|
|
|
self._jwrite = self._jwrite.outputMode(outputMode)
|
|
|
|
return self
|
|
|
|
|
|
|
|
@since(2.0)
|
|
|
|
def format(self, source):
|
|
|
|
"""Specifies the underlying output data source.
|
|
|
|
|
2017-05-26 16:33:23 -04:00
|
|
|
.. note:: Evolving.
|
2016-06-29 01:07:11 -04:00
|
|
|
|
2016-08-30 06:19:45 -04:00
|
|
|
:param source: string, name of the data source, which for now can be 'parquet'.
|
2016-06-29 01:07:11 -04:00
|
|
|
|
|
|
|
>>> writer = sdf.writeStream.format('json')
|
|
|
|
"""
|
|
|
|
self._jwrite = self._jwrite.format(source)
|
|
|
|
return self
|
|
|
|
|
|
|
|
@since(2.0)
|
|
|
|
def option(self, key, value):
|
|
|
|
"""Adds an output option for the underlying data source.
|
|
|
|
|
2017-03-15 01:30:16 -04:00
|
|
|
You can set the following option(s) for writing files:
|
|
|
|
* ``timeZone``: sets the string that indicates a timezone to be used to format
|
|
|
|
timestamps in the JSON/CSV datasources or partition values.
|
|
|
|
If it isn't set, it uses the default value, session local timezone.
|
|
|
|
|
2017-05-26 16:33:23 -04:00
|
|
|
.. note:: Evolving.
|
2016-06-29 01:07:11 -04:00
|
|
|
"""
|
|
|
|
self._jwrite = self._jwrite.option(key, to_str(value))
|
|
|
|
return self
|
|
|
|
|
|
|
|
@since(2.0)
|
|
|
|
def options(self, **options):
|
|
|
|
"""Adds output options for the underlying data source.
|
|
|
|
|
2017-03-15 01:30:16 -04:00
|
|
|
You can set the following option(s) for writing files:
|
|
|
|
* ``timeZone``: sets the string that indicates a timezone to be used to format
|
|
|
|
timestamps in the JSON/CSV datasources or partition values.
|
|
|
|
If it isn't set, it uses the default value, session local timezone.
|
|
|
|
|
2017-05-26 16:33:23 -04:00
|
|
|
.. note:: Evolving.
|
2016-06-29 01:07:11 -04:00
|
|
|
"""
|
|
|
|
for k in options:
|
|
|
|
self._jwrite = self._jwrite.option(k, to_str(options[k]))
|
|
|
|
return self
|
|
|
|
|
|
|
|
@since(2.0)
|
|
|
|
def partitionBy(self, *cols):
|
|
|
|
"""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.
|
|
|
|
|
2017-05-26 16:33:23 -04:00
|
|
|
.. note:: Evolving.
|
2016-06-29 01:07:11 -04:00
|
|
|
|
|
|
|
:param cols: name of columns
|
|
|
|
|
|
|
|
"""
|
|
|
|
if len(cols) == 1 and isinstance(cols[0], (list, tuple)):
|
|
|
|
cols = cols[0]
|
|
|
|
self._jwrite = self._jwrite.partitionBy(_to_seq(self._spark._sc, cols))
|
|
|
|
return self
|
|
|
|
|
|
|
|
@since(2.0)
|
|
|
|
def queryName(self, queryName):
|
|
|
|
"""Specifies the name of the :class:`StreamingQuery` that can be started with
|
|
|
|
:func:`start`. This name must be unique among all the currently active queries
|
|
|
|
in the associated SparkSession.
|
|
|
|
|
2017-05-26 16:33:23 -04:00
|
|
|
.. note:: Evolving.
|
2016-06-29 01:07:11 -04:00
|
|
|
|
|
|
|
:param queryName: unique name for the query
|
|
|
|
|
|
|
|
>>> writer = sdf.writeStream.queryName('streaming_query')
|
|
|
|
"""
|
|
|
|
if not queryName or type(queryName) != str or len(queryName.strip()) == 0:
|
|
|
|
raise ValueError('The queryName must be a non-empty string. Got: %s' % queryName)
|
|
|
|
self._jwrite = self._jwrite.queryName(queryName)
|
|
|
|
return self
|
|
|
|
|
|
|
|
@keyword_only
|
|
|
|
@since(2.0)
|
2018-01-18 15:25:52 -05:00
|
|
|
def trigger(self, processingTime=None, once=None, continuous=None):
|
2016-06-29 01:07:11 -04:00
|
|
|
"""Set the trigger for the stream query. If this is not set it will run the query as fast
|
|
|
|
as possible, which is equivalent to setting the trigger to ``processingTime='0 seconds'``.
|
|
|
|
|
2017-05-26 16:33:23 -04:00
|
|
|
.. note:: Evolving.
|
2016-06-29 01:07:11 -04:00
|
|
|
|
|
|
|
:param processingTime: a processing time interval as a string, e.g. '5 seconds', '1 minute'.
|
2018-01-04 00:43:14 -05:00
|
|
|
Set a trigger that runs a query periodically based on the processing
|
|
|
|
time. Only one trigger can be set.
|
|
|
|
:param once: if set to True, set a trigger that processes only one batch of data in a
|
|
|
|
streaming query then terminates the query. Only one trigger can be set.
|
2016-06-29 01:07:11 -04:00
|
|
|
|
|
|
|
>>> # trigger the query for execution every 5 seconds
|
|
|
|
>>> writer = sdf.writeStream.trigger(processingTime='5 seconds')
|
[SPARK-19876][SS][WIP] OneTime Trigger Executor
## What changes were proposed in this pull request?
An additional trigger and trigger executor that will execute a single trigger only. One can use this OneTime trigger to have more control over the scheduling of triggers.
In addition, this patch requires an optimization to StreamExecution that logs a commit record at the end of successfully processing a batch. This new commit log will be used to determine the next batch (offsets) to process after a restart, instead of using the offset log itself to determine what batch to process next after restart; using the offset log to determine this would process the previously logged batch, always, thus not permitting a OneTime trigger feature.
## How was this patch tested?
A number of existing tests have been revised. These tests all assumed that when restarting a stream, the last batch in the offset log is to be re-processed. Given that we now have a commit log that will tell us if that last batch was processed successfully, the results/assumptions of those tests needed to be revised accordingly.
In addition, a OneTime trigger test was added to StreamingQuerySuite, which tests:
- The semantics of OneTime trigger (i.e., on start, execute a single batch, then stop).
- The case when the commit log was not able to successfully log the completion of a batch before restart, which would mean that we should fall back to what's in the offset log.
- A OneTime trigger execution that results in an exception being thrown.
marmbrus tdas zsxwing
Please review http://spark.apache.org/contributing.html before opening a pull request.
Author: Tyson Condie <tcondie@gmail.com>
Author: Tathagata Das <tathagata.das1565@gmail.com>
Closes #17219 from tcondie/stream-commit.
2017-03-23 17:32:05 -04:00
|
|
|
>>> # trigger the query for just once batch of data
|
|
|
|
>>> writer = sdf.writeStream.trigger(once=True)
|
2018-01-18 15:25:52 -05:00
|
|
|
>>> # trigger the query for execution every 5 seconds
|
|
|
|
>>> writer = sdf.writeStream.trigger(continuous='5 seconds')
|
2016-06-29 01:07:11 -04:00
|
|
|
"""
|
2018-01-18 15:25:52 -05:00
|
|
|
params = [processingTime, once, continuous]
|
|
|
|
|
|
|
|
if params.count(None) == 3:
|
|
|
|
raise ValueError('No trigger provided')
|
|
|
|
elif params.count(None) < 2:
|
|
|
|
raise ValueError('Multiple triggers not allowed.')
|
|
|
|
|
[SPARK-19876][SS][WIP] OneTime Trigger Executor
## What changes were proposed in this pull request?
An additional trigger and trigger executor that will execute a single trigger only. One can use this OneTime trigger to have more control over the scheduling of triggers.
In addition, this patch requires an optimization to StreamExecution that logs a commit record at the end of successfully processing a batch. This new commit log will be used to determine the next batch (offsets) to process after a restart, instead of using the offset log itself to determine what batch to process next after restart; using the offset log to determine this would process the previously logged batch, always, thus not permitting a OneTime trigger feature.
## How was this patch tested?
A number of existing tests have been revised. These tests all assumed that when restarting a stream, the last batch in the offset log is to be re-processed. Given that we now have a commit log that will tell us if that last batch was processed successfully, the results/assumptions of those tests needed to be revised accordingly.
In addition, a OneTime trigger test was added to StreamingQuerySuite, which tests:
- The semantics of OneTime trigger (i.e., on start, execute a single batch, then stop).
- The case when the commit log was not able to successfully log the completion of a batch before restart, which would mean that we should fall back to what's in the offset log.
- A OneTime trigger execution that results in an exception being thrown.
marmbrus tdas zsxwing
Please review http://spark.apache.org/contributing.html before opening a pull request.
Author: Tyson Condie <tcondie@gmail.com>
Author: Tathagata Das <tathagata.das1565@gmail.com>
Closes #17219 from tcondie/stream-commit.
2017-03-23 17:32:05 -04:00
|
|
|
jTrigger = None
|
2016-06-29 01:07:11 -04:00
|
|
|
if processingTime is not None:
|
|
|
|
if type(processingTime) != str or len(processingTime.strip()) == 0:
|
[SPARK-19876][SS][WIP] OneTime Trigger Executor
## What changes were proposed in this pull request?
An additional trigger and trigger executor that will execute a single trigger only. One can use this OneTime trigger to have more control over the scheduling of triggers.
In addition, this patch requires an optimization to StreamExecution that logs a commit record at the end of successfully processing a batch. This new commit log will be used to determine the next batch (offsets) to process after a restart, instead of using the offset log itself to determine what batch to process next after restart; using the offset log to determine this would process the previously logged batch, always, thus not permitting a OneTime trigger feature.
## How was this patch tested?
A number of existing tests have been revised. These tests all assumed that when restarting a stream, the last batch in the offset log is to be re-processed. Given that we now have a commit log that will tell us if that last batch was processed successfully, the results/assumptions of those tests needed to be revised accordingly.
In addition, a OneTime trigger test was added to StreamingQuerySuite, which tests:
- The semantics of OneTime trigger (i.e., on start, execute a single batch, then stop).
- The case when the commit log was not able to successfully log the completion of a batch before restart, which would mean that we should fall back to what's in the offset log.
- A OneTime trigger execution that results in an exception being thrown.
marmbrus tdas zsxwing
Please review http://spark.apache.org/contributing.html before opening a pull request.
Author: Tyson Condie <tcondie@gmail.com>
Author: Tathagata Das <tathagata.das1565@gmail.com>
Closes #17219 from tcondie/stream-commit.
2017-03-23 17:32:05 -04:00
|
|
|
raise ValueError('Value for processingTime must be a non empty string. Got: %s' %
|
2016-06-29 01:07:11 -04:00
|
|
|
processingTime)
|
[SPARK-19876][SS][WIP] OneTime Trigger Executor
## What changes were proposed in this pull request?
An additional trigger and trigger executor that will execute a single trigger only. One can use this OneTime trigger to have more control over the scheduling of triggers.
In addition, this patch requires an optimization to StreamExecution that logs a commit record at the end of successfully processing a batch. This new commit log will be used to determine the next batch (offsets) to process after a restart, instead of using the offset log itself to determine what batch to process next after restart; using the offset log to determine this would process the previously logged batch, always, thus not permitting a OneTime trigger feature.
## How was this patch tested?
A number of existing tests have been revised. These tests all assumed that when restarting a stream, the last batch in the offset log is to be re-processed. Given that we now have a commit log that will tell us if that last batch was processed successfully, the results/assumptions of those tests needed to be revised accordingly.
In addition, a OneTime trigger test was added to StreamingQuerySuite, which tests:
- The semantics of OneTime trigger (i.e., on start, execute a single batch, then stop).
- The case when the commit log was not able to successfully log the completion of a batch before restart, which would mean that we should fall back to what's in the offset log.
- A OneTime trigger execution that results in an exception being thrown.
marmbrus tdas zsxwing
Please review http://spark.apache.org/contributing.html before opening a pull request.
Author: Tyson Condie <tcondie@gmail.com>
Author: Tathagata Das <tathagata.das1565@gmail.com>
Closes #17219 from tcondie/stream-commit.
2017-03-23 17:32:05 -04:00
|
|
|
interval = processingTime.strip()
|
|
|
|
jTrigger = self._spark._sc._jvm.org.apache.spark.sql.streaming.Trigger.ProcessingTime(
|
|
|
|
interval)
|
2018-01-18 15:25:52 -05:00
|
|
|
|
[SPARK-19876][SS][WIP] OneTime Trigger Executor
## What changes were proposed in this pull request?
An additional trigger and trigger executor that will execute a single trigger only. One can use this OneTime trigger to have more control over the scheduling of triggers.
In addition, this patch requires an optimization to StreamExecution that logs a commit record at the end of successfully processing a batch. This new commit log will be used to determine the next batch (offsets) to process after a restart, instead of using the offset log itself to determine what batch to process next after restart; using the offset log to determine this would process the previously logged batch, always, thus not permitting a OneTime trigger feature.
## How was this patch tested?
A number of existing tests have been revised. These tests all assumed that when restarting a stream, the last batch in the offset log is to be re-processed. Given that we now have a commit log that will tell us if that last batch was processed successfully, the results/assumptions of those tests needed to be revised accordingly.
In addition, a OneTime trigger test was added to StreamingQuerySuite, which tests:
- The semantics of OneTime trigger (i.e., on start, execute a single batch, then stop).
- The case when the commit log was not able to successfully log the completion of a batch before restart, which would mean that we should fall back to what's in the offset log.
- A OneTime trigger execution that results in an exception being thrown.
marmbrus tdas zsxwing
Please review http://spark.apache.org/contributing.html before opening a pull request.
Author: Tyson Condie <tcondie@gmail.com>
Author: Tathagata Das <tathagata.das1565@gmail.com>
Closes #17219 from tcondie/stream-commit.
2017-03-23 17:32:05 -04:00
|
|
|
elif once is not None:
|
|
|
|
if once is not True:
|
|
|
|
raise ValueError('Value for once must be True. Got: %s' % once)
|
|
|
|
jTrigger = self._spark._sc._jvm.org.apache.spark.sql.streaming.Trigger.Once()
|
2018-01-18 15:25:52 -05:00
|
|
|
|
[SPARK-19876][SS][WIP] OneTime Trigger Executor
## What changes were proposed in this pull request?
An additional trigger and trigger executor that will execute a single trigger only. One can use this OneTime trigger to have more control over the scheduling of triggers.
In addition, this patch requires an optimization to StreamExecution that logs a commit record at the end of successfully processing a batch. This new commit log will be used to determine the next batch (offsets) to process after a restart, instead of using the offset log itself to determine what batch to process next after restart; using the offset log to determine this would process the previously logged batch, always, thus not permitting a OneTime trigger feature.
## How was this patch tested?
A number of existing tests have been revised. These tests all assumed that when restarting a stream, the last batch in the offset log is to be re-processed. Given that we now have a commit log that will tell us if that last batch was processed successfully, the results/assumptions of those tests needed to be revised accordingly.
In addition, a OneTime trigger test was added to StreamingQuerySuite, which tests:
- The semantics of OneTime trigger (i.e., on start, execute a single batch, then stop).
- The case when the commit log was not able to successfully log the completion of a batch before restart, which would mean that we should fall back to what's in the offset log.
- A OneTime trigger execution that results in an exception being thrown.
marmbrus tdas zsxwing
Please review http://spark.apache.org/contributing.html before opening a pull request.
Author: Tyson Condie <tcondie@gmail.com>
Author: Tathagata Das <tathagata.das1565@gmail.com>
Closes #17219 from tcondie/stream-commit.
2017-03-23 17:32:05 -04:00
|
|
|
else:
|
2018-01-18 15:25:52 -05:00
|
|
|
if type(continuous) != str or len(continuous.strip()) == 0:
|
|
|
|
raise ValueError('Value for continuous must be a non empty string. Got: %s' %
|
|
|
|
continuous)
|
|
|
|
interval = continuous.strip()
|
|
|
|
jTrigger = self._spark._sc._jvm.org.apache.spark.sql.streaming.Trigger.Continuous(
|
|
|
|
interval)
|
|
|
|
|
[SPARK-19876][SS][WIP] OneTime Trigger Executor
## What changes were proposed in this pull request?
An additional trigger and trigger executor that will execute a single trigger only. One can use this OneTime trigger to have more control over the scheduling of triggers.
In addition, this patch requires an optimization to StreamExecution that logs a commit record at the end of successfully processing a batch. This new commit log will be used to determine the next batch (offsets) to process after a restart, instead of using the offset log itself to determine what batch to process next after restart; using the offset log to determine this would process the previously logged batch, always, thus not permitting a OneTime trigger feature.
## How was this patch tested?
A number of existing tests have been revised. These tests all assumed that when restarting a stream, the last batch in the offset log is to be re-processed. Given that we now have a commit log that will tell us if that last batch was processed successfully, the results/assumptions of those tests needed to be revised accordingly.
In addition, a OneTime trigger test was added to StreamingQuerySuite, which tests:
- The semantics of OneTime trigger (i.e., on start, execute a single batch, then stop).
- The case when the commit log was not able to successfully log the completion of a batch before restart, which would mean that we should fall back to what's in the offset log.
- A OneTime trigger execution that results in an exception being thrown.
marmbrus tdas zsxwing
Please review http://spark.apache.org/contributing.html before opening a pull request.
Author: Tyson Condie <tcondie@gmail.com>
Author: Tathagata Das <tathagata.das1565@gmail.com>
Closes #17219 from tcondie/stream-commit.
2017-03-23 17:32:05 -04:00
|
|
|
self._jwrite = self._jwrite.trigger(jTrigger)
|
2016-06-29 01:07:11 -04:00
|
|
|
return self
|
|
|
|
|
|
|
|
@ignore_unicode_prefix
|
|
|
|
@since(2.0)
|
2017-01-10 20:58:11 -05:00
|
|
|
def start(self, path=None, format=None, outputMode=None, partitionBy=None, queryName=None,
|
|
|
|
**options):
|
2016-06-29 01:07:11 -04:00
|
|
|
"""Streams the contents of the :class:`DataFrame` to a data source.
|
|
|
|
|
|
|
|
The data source is specified by the ``format`` and a set of ``options``.
|
|
|
|
If ``format`` is not specified, the default data source configured by
|
|
|
|
``spark.sql.sources.default`` will be used.
|
|
|
|
|
2017-05-26 16:33:23 -04:00
|
|
|
.. note:: Evolving.
|
2016-06-29 01:07:11 -04:00
|
|
|
|
|
|
|
:param path: the path in a Hadoop supported file system
|
|
|
|
:param format: the format used to save
|
2017-01-10 20:58:11 -05:00
|
|
|
:param outputMode: 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
|
|
|
|
* `update`:only the rows that were updated in the streaming DataFrame/Dataset will be
|
|
|
|
written to the sink every time there are some updates. If the query doesn't contain
|
|
|
|
aggregations, it will be equivalent to `append` mode.
|
2016-06-29 01:07:11 -04:00
|
|
|
:param partitionBy: names of partitioning columns
|
|
|
|
:param queryName: unique name for the query
|
|
|
|
:param options: All other string options. You may want to provide a `checkpointLocation`
|
2017-01-10 20:58:11 -05:00
|
|
|
for most streams, however it is not required for a `memory` stream.
|
2016-06-29 01:07:11 -04:00
|
|
|
|
|
|
|
>>> sq = sdf.writeStream.format('memory').queryName('this_query').start()
|
|
|
|
>>> sq.isActive
|
|
|
|
True
|
|
|
|
>>> sq.name
|
|
|
|
u'this_query'
|
|
|
|
>>> sq.stop()
|
|
|
|
>>> sq.isActive
|
|
|
|
False
|
|
|
|
>>> sq = sdf.writeStream.trigger(processingTime='5 seconds').start(
|
2017-01-10 20:58:11 -05:00
|
|
|
... queryName='that_query', outputMode="append", format='memory')
|
2016-06-29 01:07:11 -04:00
|
|
|
>>> sq.name
|
|
|
|
u'that_query'
|
|
|
|
>>> sq.isActive
|
|
|
|
True
|
|
|
|
>>> sq.stop()
|
|
|
|
"""
|
|
|
|
self.options(**options)
|
2017-01-10 20:58:11 -05:00
|
|
|
if outputMode is not None:
|
|
|
|
self.outputMode(outputMode)
|
2016-06-29 01:07:11 -04:00
|
|
|
if partitionBy is not None:
|
|
|
|
self.partitionBy(partitionBy)
|
|
|
|
if format is not None:
|
|
|
|
self.format(format)
|
|
|
|
if queryName is not None:
|
|
|
|
self.queryName(queryName)
|
|
|
|
if path is None:
|
|
|
|
return self._sq(self._jwrite.start())
|
|
|
|
else:
|
|
|
|
return self._sq(self._jwrite.start(path))
|
|
|
|
|
|
|
|
|
2016-04-28 18:22:28 -04:00
|
|
|
def _test():
|
|
|
|
import doctest
|
|
|
|
import os
|
|
|
|
import tempfile
|
2016-06-14 05:12:29 -04:00
|
|
|
from pyspark.sql import Row, SparkSession, SQLContext
|
|
|
|
import pyspark.sql.streaming
|
2016-04-28 18:22:28 -04:00
|
|
|
|
|
|
|
os.chdir(os.environ["SPARK_HOME"])
|
|
|
|
|
2016-06-14 05:12:29 -04:00
|
|
|
globs = pyspark.sql.streaming.__dict__.copy()
|
|
|
|
try:
|
2016-06-14 20:58:45 -04:00
|
|
|
spark = SparkSession.builder.getOrCreate()
|
2016-06-14 05:12:29 -04:00
|
|
|
except py4j.protocol.Py4JError:
|
|
|
|
spark = SparkSession(sc)
|
2016-04-28 18:22:28 -04:00
|
|
|
|
|
|
|
globs['tempfile'] = tempfile
|
|
|
|
globs['os'] = os
|
2016-06-14 05:12:29 -04:00
|
|
|
globs['spark'] = spark
|
|
|
|
globs['sqlContext'] = SQLContext.getOrCreate(spark.sparkContext)
|
2016-06-29 01:07:11 -04:00
|
|
|
globs['sdf'] = \
|
|
|
|
spark.readStream.format('text').load('python/test_support/sql/streaming')
|
|
|
|
globs['sdf_schema'] = StructType([StructField("data", StringType(), False)])
|
2016-04-28 18:22:28 -04:00
|
|
|
globs['df'] = \
|
2016-06-14 20:58:45 -04:00
|
|
|
globs['spark'].readStream.format('text').load('python/test_support/sql/streaming')
|
2016-04-28 18:22:28 -04:00
|
|
|
|
|
|
|
(failure_count, test_count) = doctest.testmod(
|
2016-06-14 05:12:29 -04:00
|
|
|
pyspark.sql.streaming, globs=globs,
|
2016-04-28 18:22:28 -04:00
|
|
|
optionflags=doctest.ELLIPSIS | doctest.NORMALIZE_WHITESPACE | doctest.REPORT_NDIFF)
|
2016-06-14 05:12:29 -04:00
|
|
|
globs['spark'].stop()
|
[SPARK-17731][SQL][STREAMING] Metrics for structured streaming
## What changes were proposed in this pull request?
Metrics are needed for monitoring structured streaming apps. Here is the design doc for implementing the necessary metrics.
https://docs.google.com/document/d/1NIdcGuR1B3WIe8t7VxLrt58TJB4DtipWEbj5I_mzJys/edit?usp=sharing
Specifically, this PR adds the following public APIs changes.
### New APIs
- `StreamingQuery.status` returns a `StreamingQueryStatus` object (renamed from `StreamingQueryInfo`, see later)
- `StreamingQueryStatus` has the following important fields
- inputRate - Current rate (rows/sec) at which data is being generated by all the sources
- processingRate - Current rate (rows/sec) at which the query is processing data from
all the sources
- ~~outputRate~~ - *Does not work with wholestage codegen*
- latency - Current average latency between the data being available in source and the sink writing the corresponding output
- sourceStatuses: Array[SourceStatus] - Current statuses of the sources
- sinkStatus: SinkStatus - Current status of the sink
- triggerStatus - Low-level detailed status of the last completed/currently active trigger
- latencies - getOffset, getBatch, full trigger, wal writes
- timestamps - trigger start, finish, after getOffset, after getBatch
- numRows - input, output, state total/updated rows for aggregations
- `SourceStatus` has the following important fields
- inputRate - Current rate (rows/sec) at which data is being generated by the source
- processingRate - Current rate (rows/sec) at which the query is processing data from the source
- triggerStatus - Low-level detailed status of the last completed/currently active trigger
- Python API for `StreamingQuery.status()`
### Breaking changes to existing APIs
**Existing direct public facing APIs**
- Deprecated direct public-facing APIs `StreamingQuery.sourceStatuses` and `StreamingQuery.sinkStatus` in favour of `StreamingQuery.status.sourceStatuses/sinkStatus`.
- Branch 2.0 should have it deprecated, master should have it removed.
**Existing advanced listener APIs**
- `StreamingQueryInfo` renamed to `StreamingQueryStatus` for consistency with `SourceStatus`, `SinkStatus`
- Earlier StreamingQueryInfo was used only in the advanced listener API, but now it is used in direct public-facing API (StreamingQuery.status)
- Field `queryInfo` in listener events `QueryStarted`, `QueryProgress`, `QueryTerminated` changed have name `queryStatus` and return type `StreamingQueryStatus`.
- Field `offsetDesc` in `SourceStatus` was Option[String], converted it to `String`.
- For `SourceStatus` and `SinkStatus` made constructor private instead of private[sql] to make them more java-safe. Instead added `private[sql] object SourceStatus/SinkStatus.apply()` which are harder to accidentally use in Java.
## How was this patch tested?
Old and new unit tests.
- Rate calculation and other internal logic of StreamMetrics tested by StreamMetricsSuite.
- New info in statuses returned through StreamingQueryListener is tested in StreamingQueryListenerSuite.
- New and old info returned through StreamingQuery.status is tested in StreamingQuerySuite.
- Source-specific tests for making sure input rows are counted are is source-specific test suites.
- Additional tests to test minor additions in LocalTableScanExec, StateStore, etc.
Metrics also manually tested using Ganglia sink
Author: Tathagata Das <tathagata.das1565@gmail.com>
Closes #15307 from tdas/SPARK-17731.
2016-10-13 16:36:26 -04:00
|
|
|
|
2016-04-28 18:22:28 -04:00
|
|
|
if failure_count:
|
2018-03-08 06:38:34 -05:00
|
|
|
sys.exit(-1)
|
2016-04-28 18:22:28 -04:00
|
|
|
|
|
|
|
|
|
|
|
if __name__ == "__main__":
|
|
|
|
_test()
|