Coverage for pyspark/streaming/context.py : 83%
Hot-keys on this page
r m x p toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
|
# # 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. #
""" Main entry point for Spark Streaming functionality. A StreamingContext represents the connection to a Spark cluster, and can be used to create :class:`DStream` various input sources. It can be from an existing :class:`SparkContext`. After creating and transforming DStreams, the streaming computation can be started and stopped using `context.start()` and `context.stop()`, respectively. `context.awaitTermination()` allows the current thread to wait for the termination of the context by `stop()` or by an exception.
Parameters ---------- sparkContext : :class:`SparkContext` SparkContext object. batchDuration : int, optional the time interval (in seconds) at which streaming data will be divided into batches """
# Reference to a currently active StreamingContext
""" Create Duration object given number of seconds """
def _ensure_initialized(cls):
# register serializer for TransformFunction # it happens before creating SparkContext when loading from checkpointing SparkContext._active_spark_context, CloudPickleSerializer(), gw)
def getOrCreate(cls, checkpointPath, setupFunc): """ Either recreate a StreamingContext from checkpoint data or create a new StreamingContext. If checkpoint data exists in the provided `checkpointPath`, then StreamingContext will be recreated from the checkpoint data. If the data does not exist, then the provided setupFunc will be used to create a new context.
Parameters ---------- checkpointPath : str Checkpoint directory used in an earlier streaming program setupFunc : function Function to create a new context and setup DStreams """
# Check whether valid checkpoint information exists in the given path
# If there is already an active instance of Python SparkContext use it, or create a new one
# update ctx in serializer
def getActive(cls): """ Return either the currently active StreamingContext (i.e., if there is a context started but not stopped) or None. """ # Verify that the current running Java StreamingContext is active and is the same one # backing the supposedly active Python context
cls._activeContext = None raise RuntimeError( "JVM's active JavaStreamingContext is not the JavaStreamingContext " "backing the action Python StreamingContext. This is unexpected.")
def getActiveOrCreate(cls, checkpointPath, setupFunc): """ Either return the active StreamingContext (i.e. currently started but not stopped), or recreate a StreamingContext from checkpoint data or create a new StreamingContext using the provided setupFunc function. If the checkpointPath is None or does not contain valid checkpoint data, then setupFunc will be called to create a new context and setup DStreams.
Parameters ---------- checkpointPath : str Checkpoint directory used in an earlier streaming program. Can be None if the intention is to always create a new context when there is no active context. setupFunc : function Function to create a new JavaStreamingContext and setup DStreams """
raise TypeError("setupFunc should be callable.") else:
def sparkContext(self): """ Return SparkContext which is associated with this StreamingContext. """
""" Start the execution of the streams. """
""" Wait for the execution to stop.
Parameters ---------- timeout : int, optional time to wait in seconds """ if timeout is None: self._jssc.awaitTermination() else: self._jssc.awaitTerminationOrTimeout(int(timeout * 1000))
""" Wait for the execution to stop. Return `true` if it's stopped; or throw the reported error during the execution; or `false` if the waiting time elapsed before returning from the method.
Parameters ---------- timeout : int time to wait in seconds """
""" Stop the execution of the streams, with option of ensuring all received data has been processed.
Parameters ---------- stopSparkContext : bool, optional Stop the associated SparkContext or not stopGracefully : bool, optional Stop gracefully by waiting for the processing of all received data to be completed """
""" Set each DStreams in this context to remember RDDs it generated in the last given duration. DStreams remember RDDs only for a limited duration of time and releases them for garbage collection. This method allows the developer to specify how long to remember the RDDs (if the developer wishes to query old data outside the DStream computation).
Parameters ---------- duration : int Minimum duration (in seconds) that each DStream should remember its RDDs """
""" Sets the context to periodically checkpoint the DStream operations for master fault-tolerance. The graph will be checkpointed every batch interval.
Parameters ---------- directory : str HDFS-compatible directory where the checkpoint data will be reliably stored """
""" Create an input from TCP source hostname:port. Data is received using a TCP socket and receive byte is interpreted as UTF8 encoded ``\\n`` delimited lines.
Parameters ---------- hostname : str Hostname to connect to for receiving data port : int Port to connect to for receiving data storageLevel : :class:`pyspark.StorageLevel`, optional Storage level to use for storing the received objects """ jlevel = self._sc._getJavaStorageLevel(storageLevel) return DStream(self._jssc.socketTextStream(hostname, port, jlevel), self, UTF8Deserializer())
""" Create an input stream that monitors a Hadoop-compatible file system for new files and reads them as text files. Files must be written to the monitored directory by "moving" them from another location within the same file system. File names starting with . are ignored. The text files must be encoded as UTF-8. """
""" Create an input stream that monitors a Hadoop-compatible file system for new files and reads them as flat binary files with records of fixed length. Files must be written to the monitored directory by "moving" them from another location within the same file system. File names starting with . are ignored.
Parameters ---------- directory : str Directory to load data from recordLength : int Length of each record in bytes """ NoOpSerializer())
# make sure they have same serializer # reset them to sc.serializer
""" Create an input stream from a queue of RDDs or list. In each batch, it will process either one or all of the RDDs returned by the queue.
Parameters ---------- rdds : list Queue of RDDs oneAtATime : bool, optional pick one rdd each time or pick all of them once. default : :class:`pyspark.RDD`, optional The default rdd if no more in rdds
Notes ----- Changes to the queue after the stream is created will not be recognized. """ default = self._sc.parallelize(default)
rdds = [rdds]
default = default._reserialize(rdds[0]._jrdd_deserializer) jdstream = self._jssc.queueStream(queue, oneAtATime, default._jrdd) else:
""" Create a new DStream in which each RDD is generated by applying a function on RDDs of the DStreams. The order of the JavaRDDs in the transform function parameter will be the same as the order of corresponding DStreams in the list. """ # change the final serializer to sc.serializer lambda t, *rdds: transformFunc(rdds), *[d._jrdd_deserializer for d in dstreams])
""" Create a unified DStream from multiple DStreams of the same type and same slide duration. """ raise ValueError("should have at least one DStream to union") return dstreams[0] raise ValueError("All DStreams should have same serializer") raise ValueError("All DStreams should have same slide duration") elif is_instance_of(gw, dstreams[0]._jdstream, jpair_dstream_cls): cls = jpair_dstream_cls else: cls_name = dstreams[0]._jdstream.getClass().getCanonicalName() raise TypeError("Unsupported Java DStream class %s" % cls_name)
""" Add a [[org.apache.spark.streaming.scheduler.StreamingListener]] object for receiving system events related to streaming. """ self._jvm.PythonStreamingListenerWrapper(streamingListener))) |