<commit_before>#
# 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.
#
import os
import sys

from py4j.java_collections import ListConverter
from py4j.java_gateway import java_import, JavaObject

from pyspark import RDD, SparkConf
from pyspark.serializers import UTF8Deserializer, CloudPickleSerializer
from pyspark.context import SparkContext
from pyspark.storagelevel import StorageLevel
from pyspark.streaming.dstream import DStream
from pyspark.streaming.util import TransformFunction, TransformFunctionSerializer

__all__ = ["StreamingContext"]


def _daemonize_callback_server():
    """
    Hack Py4J to daemonize callback server

    The thread of callback server has daemon=False, it will block the driver
    from exiting if it's not shutdown. The following code replace `start()`
    of CallbackServer with a new version, which set daemon=True for this
    thread.

    Also, it will update the port number (0) with real port
    """
    # TODO: create a patch for Py4J
    import socket
    import py4j.java_gateway
    logger = py4j.java_gateway.logger
    from py4j.java_gateway import Py4JNetworkError
    from threading import Thread

    def start(self):
        """Starts the CallbackServer. This method should be called by the
        client instead of run()."""
        self.server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR,
                                      1)
        try:
            self.server_socket.bind((self.address, self.port))
            if not self.port:
                # update port with real port
                self.port = self.server_socket.getsockname()[1]
        except Exception as e:
            msg = 'An error occurred while trying to start the callback server: %s' % e
            logger.exception(msg)
            raise Py4JNetworkError(msg)

        # Maybe thread needs to be cleanup up?
        self.thread = Thread(target=self.run)
        self.thread.daemon = True
        self.thread.start()

    py4j.java_gateway.CallbackServer.start = start


class StreamingContext(object):
    """
    Main entry point for Spark Streaming functionality. A StreamingContext
    represents the connection to a Spark cluster, and can be used to create
    L{DStream} various input sources. It can be from an existing L{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.
    """
    _transformerSerializer = None

    def __init__(self, sparkContext, batchDuration=None, jssc=None):
        """
        Create a new StreamingContext.

        @param sparkContext: L{SparkContext} object.
        @param batchDuration: the time interval (in seconds) at which streaming
                              data will be divided into batches
        """

        self._sc = sparkContext
        self._jvm = self._sc._jvm
        self._jssc = jssc or self._initialize_context(self._sc, batchDuration)

    def _initialize_context(self, sc, duration):
        self._ensure_initialized()
        return self._jvm.JavaStreamingContext(sc._jsc, self._jduration(duration))

    def _jduration(self, seconds):
        """
        Create Duration object given number of seconds
        """
        return self._jvm.Duration(int(seconds * 1000))

    @classmethod
    def _ensure_initialized(cls):
        SparkContext._ensure_initialized()
        gw = SparkContext._gateway

        java_import(gw.jvm, "org.apache.spark.streaming.*")
        java_import(gw.jvm, "org.apache.spark.streaming.api.java.*")
        java_import(gw.jvm, "org.apache.spark.streaming.api.python.*")

        # start callback server
        # getattr will fallback to JVM, so we cannot test by hasattr()
        if "_callback_server" not in gw.__dict__:
            _daemonize_callback_server()
            # use random port
            gw._start_callback_server(0)
            # gateway with real port
            gw._python_proxy_port = gw._callback_server.port
            # get the GatewayServer object in JVM by ID
            jgws = JavaObject("GATEWAY_SERVER", gw._gateway_client)
            # update the port of CallbackClient with real port
            gw.jvm.PythonDStream.updatePythonGatewayPort(jgws, gw._python_proxy_port)

        # register serializer for TransformFunction
        # it happens before creating SparkContext when loading from checkpointing
        cls._transformerSerializer = TransformFunctionSerializer(
            SparkContext._active_spark_context, CloudPickleSerializer(), gw)

    @classmethod
    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 JavaStreamingContext.

        @param checkpointPath Checkpoint directory used in an earlier JavaStreamingContext program
        @param setupFunc      Function to create a new JavaStreamingContext and setup DStreams
        """
        # TODO: support checkpoint in HDFS
        if not os.path.exists(checkpointPath) or not os.listdir(checkpointPath):
            ssc = setupFunc()
            ssc.checkpoint(checkpointPath)
            return ssc

        cls._ensure_initialized()
        gw = SparkContext._gateway

        try:
            jssc = gw.jvm.JavaStreamingContext(checkpointPath)
        except Exception:
            print >>sys.stderr, "failed to load StreamingContext from checkpoint"
            raise

        jsc = jssc.sparkContext()
        conf = SparkConf(_jconf=jsc.getConf())
        sc = SparkContext(conf=conf, gateway=gw, jsc=jsc)<commit_msg>[DOC][PySpark][Streaming] Fix docstring for sphinx<commit_after>#
# 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.
#
import os
import sys

from py4j.java_collections import ListConverter
from py4j.java_gateway import java_import, JavaObject

from pyspark import RDD, SparkConf
from pyspark.serializers import UTF8Deserializer, CloudPickleSerializer
from pyspark.context import SparkContext
from pyspark.storagelevel import StorageLevel
from pyspark.streaming.dstream import DStream
from pyspark.streaming.util import TransformFunction, TransformFunctionSerializer

__all__ = ["StreamingContext"]


def _daemonize_callback_server():
    """
    Hack Py4J to daemonize callback server

    The thread of callback server has daemon=False, it will block the driver
    from exiting if it's not shutdown. The following code replace `start()`
    of CallbackServer with a new version, which set daemon=True for this
    thread.

    Also, it will update the port number (0) with real port
    """
    # TODO: create a patch for Py4J
    import socket
    import py4j.java_gateway
    logger = py4j.java_gateway.logger
    from py4j.java_gateway import Py4JNetworkError
    from threading import Thread

    def start(self):
        """Starts the CallbackServer. This method should be called by the
        client instead of run()."""
        self.server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR,
                                      1)
        try:
            self.server_socket.bind((self.address, self.port))
            if not self.port:
                # update port with real port
                self.port = self.server_socket.getsockname()[1]
        except Exception as e:
            msg = 'An error occurred while trying to start the callback server: %s' % e
            logger.exception(msg)
            raise Py4JNetworkError(msg)

        # Maybe thread needs to be cleanup up?
        self.thread = Thread(target=self.run)
        self.thread.daemon = True
        self.thread.start()

    py4j.java_gateway.CallbackServer.start = start


class StreamingContext(object):
    """
    Main entry point for Spark Streaming functionality. A StreamingContext
    represents the connection to a Spark cluster, and can be used to create
    L{DStream} various input sources. It can be from an existing L{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.
    """
    _transformerSerializer = None

    def __init__(self, sparkContext, batchDuration=None, jssc=None):
        """
        Create a new StreamingContext.

        @param sparkContext: L{SparkContext} object.
        @param batchDuration: the time interval (in seconds) at which streaming
                              data will be divided into batches
        """

        self._sc = sparkContext
        self._jvm = self._sc._jvm
        self._jssc = jssc or self._initialize_context(self._sc, batchDuration)

    def _initialize_context(self, sc, duration):
        self._ensure_initialized()
        return self._jvm.JavaStreamingContext(sc._jsc, self._jduration(duration))

    def _jduration(self, seconds):
        """
        Create Duration object given number of seconds
        """
        return self._jvm.Duration(int(seconds * 1000))

    @classmethod
    def _ensure_initialized(cls):
        SparkContext._ensure_initialized()
        gw = SparkContext._gateway

        java_import(gw.jvm, "org.apache.spark.streaming.*")
        java_import(gw.jvm, "org.apache.spark.streaming.api.java.*")
        java_import(gw.jvm, "org.apache.spark.streaming.api.python.*")

        # start callback server
        # getattr will fallback to JVM, so we cannot test by hasattr()
        if "_callback_server" not in gw.__dict__:
            _daemonize_callback_server()
            # use random port
            gw._start_callback_server(0)
            # gateway with real port
            gw._python_proxy_port = gw._callback_server.port
            # get the GatewayServer object in JVM by ID
            jgws = JavaObject("GATEWAY_SERVER", gw._gateway_client)
            # update the port of CallbackClient with real port
            gw.jvm.PythonDStream.updatePythonGatewayPort(jgws, gw._python_proxy_port)

        # register serializer for TransformFunction
        # it happens before creating SparkContext when loading from checkpointing
        cls._transformerSerializer = TransformFunctionSerializer(
            SparkContext._active_spark_context, CloudPickleSerializer(), gw)

    @classmethod
    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 JavaStreamingContext.

        @param checkpointPath: Checkpoint directory used in an earlier JavaStreamingContext program
        @param setupFunc:      Function to create a new JavaStreamingContext and setup DStreams
        """
        # TODO: support checkpoint in HDFS
        if not os.path.exists(checkpointPath) or not os.listdir(checkpointPath):
            ssc = setupFunc()
            ssc.checkpoint(checkpointPath)
            return ssc

        cls._ensure_initialized()
        gw = SparkContext._gateway

        try:
            jssc = gw.jvm.JavaStreamingContext(checkpointPath)
        except Exception:
            print >>sys.stderr, "failed to load StreamingContext from checkpoint"
            raise

        jsc = jssc.sparkContext()
        conf = SparkConf(_jconf=jsc.getConf())
        sc = SparkContext(conf=conf, gateway=gw, jsc=jsc)
{'fraction_non_alphanumeric': 0.06743553955777566, 'fraction_numerical': 0.0033791228972305885, 'mean_word_length': 2.886383100199829, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5697554', 'n_tokens_mistral': 3694, 'n_tokens_neox': 3549, 'n_words': 1255}