org.apache.spark.streaming

StreamingContext

class StreamingContext extends Logging

A StreamingContext is the main entry point for Spark Streaming functionality. Besides the basic information (such as, cluster URL and job name) to internally create a SparkContext, it provides methods used to create DStream from various input sources.

Linear Supertypes
Logging, AnyRef, Any
Ordering
  1. Alphabetic
  2. By inheritance
Inherited
  1. Hide All
  2. Show all
  1. StreamingContext
  2. Logging
  3. AnyRef
  4. Any
Visibility
  1. Public
  2. All

Instance Constructors

  1. new StreamingContext(path: String)

    Re-create a StreamingContext from a checkpoint file.

    Re-create a StreamingContext from a checkpoint file.

    path

    Path either to the directory that was specified as the checkpoint directory, or to the checkpoint file 'graph' or 'graph.bk'.

  2. new StreamingContext(master: String, appName: String, batchDuration: Duration, sparkHome: String = null, jars: Seq[String] = Nil, environment: Map[String, String] = Map())

    Create a StreamingContext by providing the details necessary for creating a new SparkContext.

    Create a StreamingContext by providing the details necessary for creating a new SparkContext.

    master

    Cluster URL to connect to (e.g. mesos://host:port, spark://host:port, local[4]).

    appName

    A name for your job, to display on the cluster web UI

    batchDuration

    The time interval at which streaming data will be divided into batches

  3. new StreamingContext(sparkContext: SparkContext, batchDuration: Duration)

    Create a StreamingContext using an existing SparkContext.

    Create a StreamingContext using an existing SparkContext.

    sparkContext

    Existing SparkContext

    batchDuration

    The time interval at which streaming data will be divided into batches

Value Members

  1. final def !=(arg0: AnyRef): Boolean

    Definition Classes
    AnyRef
  2. final def !=(arg0: Any): Boolean

    Definition Classes
    Any
  3. final def ##(): Int

    Definition Classes
    AnyRef → Any
  4. final def ==(arg0: AnyRef): Boolean

    Definition Classes
    AnyRef
  5. final def ==(arg0: Any): Boolean

    Definition Classes
    Any
  6. def actorStream[T](props: Props, name: String, storageLevel: StorageLevel = StorageLevel.MEMORY_ONLY_SER_2, supervisorStrategy: SupervisorStrategy = ...)(implicit arg0: ClassManifest[T]): DStream[T]

    Create an input stream with any arbitrary user implemented actor receiver.

    Create an input stream with any arbitrary user implemented actor receiver. Find more details at: http://spark-project.org/docs/latest/streaming-custom-receivers.html

    props

    Props object defining creation of the actor

    name

    Name of the actor

    storageLevel

    RDD storage level. Defaults to memory-only.

    Note

    An important point to note: Since Actor may exist outside the spark framework, It is thus user's responsibility to ensure the type safety, i.e parametrized type of data received and actorStream should be same.

  7. final def asInstanceOf[T0]: T0

    Definition Classes
    Any
  8. def checkpoint(directory: String): Unit

    Set the context to periodically checkpoint the DStream operations for master fault-tolerance.

    Set the context to periodically checkpoint the DStream operations for master fault-tolerance. The graph will be checkpointed every batch interval.

    directory

    HDFS-compatible directory where the checkpoint data will be reliably stored

  9. var checkpointDir: String

    Attributes
    protected[streaming]
  10. var checkpointDuration: Duration

    Attributes
    protected[streaming]
  11. def clone(): AnyRef

    Attributes
    protected[lang]
    Definition Classes
    AnyRef
    Annotations
    @throws()
  12. val env: SparkEnv

    Attributes
    protected[streaming]
  13. final def eq(arg0: AnyRef): Boolean

    Definition Classes
    AnyRef
  14. def equals(arg0: Any): Boolean

    Definition Classes
    AnyRef → Any
  15. def fileStream[K, V, F <: InputFormat[K, V]](directory: String, filter: (Path) ⇒ Boolean, newFilesOnly: Boolean)(implicit arg0: ClassManifest[K], arg1: ClassManifest[V], arg2: ClassManifest[F]): DStream[(K, V)]

    Create a input stream that monitors a Hadoop-compatible filesystem for new files and reads them using the given key-value types and input format.

    Create a input stream that monitors a Hadoop-compatible filesystem for new files and reads them using the given key-value types and input format.

    K

    Key type for reading HDFS file

    V

    Value type for reading HDFS file

    F

    Input format for reading HDFS file

    directory

    HDFS directory to monitor for new file

    filter

    Function to filter paths to process

    newFilesOnly

    Should process only new files and ignore existing files in the directory

  16. def fileStream[K, V, F <: InputFormat[K, V]](directory: String)(implicit arg0: ClassManifest[K], arg1: ClassManifest[V], arg2: ClassManifest[F]): DStream[(K, V)]

    Create a input stream that monitors a Hadoop-compatible filesystem for new files and reads them using the given key-value types and input format.

    Create a input stream that monitors a Hadoop-compatible filesystem for new files and reads them using the given key-value types and input format. File names starting with . are ignored.

    K

    Key type for reading HDFS file

    V

    Value type for reading HDFS file

    F

    Input format for reading HDFS file

    directory

    HDFS directory to monitor for new file

  17. def finalize(): Unit

    Attributes
    protected[lang]
    Definition Classes
    AnyRef
    Annotations
    @throws()
  18. def flumeStream(hostname: String, port: Int, storageLevel: StorageLevel = StorageLevel.MEMORY_AND_DISK_SER_2): DStream[SparkFlumeEvent]

    Create a input stream from a Flume source.

    Create a input stream from a Flume source.

    hostname

    Hostname of the slave machine to which the flume data will be sent

    port

    Port of the slave machine to which the flume data will be sent

    storageLevel

    Storage level to use for storing the received objects

  19. final def getClass(): java.lang.Class[_]

    Definition Classes
    AnyRef → Any
  20. def getNewNetworkStreamId(): Int

    Attributes
    protected[streaming]
  21. val graph: DStreamGraph

    Attributes
    protected[streaming]
  22. def hashCode(): Int

    Definition Classes
    AnyRef → Any
  23. def initLogging(): Unit

    Attributes
    protected
    Definition Classes
    Logging
  24. def initialCheckpoint: Checkpoint

    Attributes
    protected[streaming]
  25. val isCheckpointPresent: Boolean

    Attributes
    protected[streaming]
  26. final def isInstanceOf[T0]: Boolean

    Definition Classes
    Any
  27. def isTraceEnabled(): Boolean

    Attributes
    protected
    Definition Classes
    Logging
  28. def kafkaStream[T, D <: kafka.serializer.Decoder[_]](kafkaParams: Map[String, String], topics: Map[String, Int], storageLevel: StorageLevel)(implicit arg0: ClassManifest[T], arg1: Manifest[D]): DStream[T]

    Create an input stream that pulls messages from a Kafka Broker.

    Create an input stream that pulls messages from a Kafka Broker.

    kafkaParams

    Map of kafka configuration paramaters. See: http://kafka.apache.org/configuration.html

    topics

    Map of (topic_name -> numPartitions) to consume. Each partition is consumed in its own thread.

    storageLevel

    Storage level to use for storing the received objects

  29. def kafkaStream(zkQuorum: String, groupId: String, topics: Map[String, Int], storageLevel: StorageLevel = StorageLevel.MEMORY_ONLY_SER_2): DStream[String]

    Create an input stream that pulls messages from a Kafka Broker.

    Create an input stream that pulls messages from a Kafka Broker.

    zkQuorum

    Zookeper quorum (hostname:port,hostname:port,..).

    groupId

    The group id for this consumer.

    topics

    Map of (topic_name -> numPartitions) to consume. Each partition is consumed in its own thread.

    storageLevel

    Storage level to use for storing the received objects (default: StorageLevel.MEMORY_AND_DISK_SER_2)

  30. def log: Logger

    Attributes
    protected
    Definition Classes
    Logging
  31. def logDebug(msg: ⇒ String, throwable: Throwable): Unit

    Attributes
    protected
    Definition Classes
    Logging
  32. def logDebug(msg: ⇒ String): Unit

    Attributes
    protected
    Definition Classes
    Logging
  33. def logError(msg: ⇒ String, throwable: Throwable): Unit

    Attributes
    protected
    Definition Classes
    Logging
  34. def logError(msg: ⇒ String): Unit

    Attributes
    protected
    Definition Classes
    Logging
  35. def logInfo(msg: ⇒ String, throwable: Throwable): Unit

    Attributes
    protected
    Definition Classes
    Logging
  36. def logInfo(msg: ⇒ String): Unit

    Attributes
    protected
    Definition Classes
    Logging
  37. def logTrace(msg: ⇒ String, throwable: Throwable): Unit

    Attributes
    protected
    Definition Classes
    Logging
  38. def logTrace(msg: ⇒ String): Unit

    Attributes
    protected
    Definition Classes
    Logging
  39. def logWarning(msg: ⇒ String, throwable: Throwable): Unit

    Attributes
    protected
    Definition Classes
    Logging
  40. def logWarning(msg: ⇒ String): Unit

    Attributes
    protected
    Definition Classes
    Logging
  41. final def ne(arg0: AnyRef): Boolean

    Definition Classes
    AnyRef
  42. var networkInputTracker: NetworkInputTracker

    Attributes
    protected[streaming]
  43. def networkStream[T](receiver: NetworkReceiver[T])(implicit arg0: ClassManifest[T]): DStream[T]

    Create an input stream with any arbitrary user implemented network receiver.

    Create an input stream with any arbitrary user implemented network receiver. Find more details at: http://spark-project.org/docs/latest/streaming-custom-receivers.html

    receiver

    Custom implementation of NetworkReceiver

  44. val nextNetworkInputStreamId: AtomicInteger

    Attributes
    protected[streaming]
  45. final def notify(): Unit

    Definition Classes
    AnyRef
  46. final def notifyAll(): Unit

    Definition Classes
    AnyRef
  47. def queueStream[T](queue: Queue[RDD[T]], oneAtATime: Boolean, defaultRDD: RDD[T])(implicit arg0: ClassManifest[T]): DStream[T]

    Create an input stream from a queue of RDDs.

    Create an input stream from a queue of RDDs. In each batch, it will process either one or all of the RDDs returned by the queue.

    T

    Type of objects in the RDD

    queue

    Queue of RDDs

    oneAtATime

    Whether only one RDD should be consumed from the queue in every interval

    defaultRDD

    Default RDD is returned by the DStream when the queue is empty. Set as null if no RDD should be returned when empty

  48. def queueStream[T](queue: Queue[RDD[T]], oneAtATime: Boolean)(implicit arg0: ClassManifest[T]): DStream[T]

    Create an input stream from a queue of RDDs.

    Create an input stream from a queue of RDDs. In each batch, it will process either one or all of the RDDs returned by the queue.

    T

    Type of objects in the RDD

    queue

    Queue of RDDs

    oneAtATime

    Whether only one RDD should be consumed from the queue in every interval

  49. def rawSocketStream[T](hostname: String, port: Int, storageLevel: StorageLevel = StorageLevel.MEMORY_AND_DISK_SER_2)(implicit arg0: ClassManifest[T]): DStream[T]

    Create a input stream from network source hostname:port, where data is received as serialized blocks (serialized using the Spark's serializer) that can be directly pushed into the block manager without deserializing them.

    Create a input stream from network source hostname:port, where data is received as serialized blocks (serialized using the Spark's serializer) that can be directly pushed into the block manager without deserializing them. This is the most efficient way to receive data.

    T

    Type of the objects in the received blocks

    hostname

    Hostname to connect to for receiving data

    port

    Port to connect to for receiving data

    storageLevel

    Storage level to use for storing the received objects

  50. var receiverJobThread: Thread

    Attributes
    protected[streaming]
  51. def registerInputStream(inputStream: org.apache.spark.streaming.dstream.InputDStream[_]): Unit

    Register an input stream that will be started (InputDStream.

    Register an input stream that will be started (InputDStream.start() called) to get the input data.

  52. def registerOutputStream(outputStream: org.apache.spark.streaming.DStream[_]): Unit

    Register an output stream that will be computed every interval

  53. def remember(duration: Duration): Unit

    Set each DStreams in this context to remember RDDs it generated in the last given duration.

    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 to long to remember the RDDs ( if the developer wishes to query old data outside the DStream computation).

    duration

    Minimum duration that each DStream should remember its RDDs

  54. val sc: SparkContext

    Attributes
    protected[streaming]
  55. var scheduler: Scheduler

    Attributes
    protected[streaming]
  56. def socketStream[T](hostname: String, port: Int, converter: (InputStream) ⇒ Iterator[T], storageLevel: StorageLevel)(implicit arg0: ClassManifest[T]): DStream[T]

    Create a input stream from TCP source hostname:port.

    Create a input stream from TCP source hostname:port. Data is received using a TCP socket and the receive bytes it interepreted as object using the given converter.

    T

    Type of the objects received (after converting bytes to objects)

    hostname

    Hostname to connect to for receiving data

    port

    Port to connect to for receiving data

    converter

    Function to convert the byte stream to objects

    storageLevel

    Storage level to use for storing the received objects

  57. def socketTextStream(hostname: String, port: Int, storageLevel: StorageLevel = StorageLevel.MEMORY_AND_DISK_SER_2): DStream[String]

    Create a input stream from TCP source hostname:port.

    Create a input stream from TCP source hostname:port. Data is received using a TCP socket and the receive bytes is interpreted as UTF8 encoded \n delimited lines.

    hostname

    Hostname to connect to for receiving data

    port

    Port to connect to for receiving data

    storageLevel

    Storage level to use for storing the received objects (default: StorageLevel.MEMORY_AND_DISK_SER_2)

  58. def sparkContext: SparkContext

    Return the associated Spark context

  59. def start(): Unit

    Start the execution of the streams.

  60. def stop(): Unit

    Stop the execution of the streams.

  61. final def synchronized[T0](arg0: ⇒ T0): T0

    Definition Classes
    AnyRef
  62. def textFileStream(directory: String): DStream[String]

    Create a input stream that monitors a Hadoop-compatible filesystem for new files and reads them as text files (using key as LongWritable, value as Text and input format as TextInputFormat).

    Create a input stream that monitors a Hadoop-compatible filesystem for new files and reads them as text files (using key as LongWritable, value as Text and input format as TextInputFormat). File names starting with . are ignored.

    directory

    HDFS directory to monitor for new file

  63. def toString(): String

    Definition Classes
    AnyRef → Any
  64. def transform[T](dstreams: Seq[org.apache.spark.streaming.DStream[_]], transformFunc: (Seq[org.apache.spark.rdd.RDD[_]], Time) ⇒ RDD[T])(implicit arg0: ClassManifest[T]): DStream[T]

    Create a new DStream in which each RDD is generated by applying a function on RDDs of the DStreams.

  65. def twitterStream(twitterAuth: Option[Authorization] = None, filters: Seq[String] = Nil, storageLevel: StorageLevel = StorageLevel.MEMORY_AND_DISK_SER_2): DStream[Status]

    Create a input stream that returns tweets received from Twitter.

    Create a input stream that returns tweets received from Twitter.

    twitterAuth

    Twitter4J authentication, or None to use Twitter4J's default OAuth authorization; this uses the system properties twitter4j.oauth.consumerKey, .consumerSecret, .accessToken and .accessTokenSecret.

    filters

    Set of filter strings to get only those tweets that match them

    storageLevel

    Storage level to use for storing the received objects

  66. def union[T](streams: Seq[DStream[T]])(implicit arg0: ClassManifest[T]): DStream[T]

    Create a unified DStream from multiple DStreams of the same type and same slide duration.

  67. def validate(): Unit

    Attributes
    protected
  68. final def wait(): Unit

    Definition Classes
    AnyRef
    Annotations
    @throws()
  69. final def wait(arg0: Long, arg1: Int): Unit

    Definition Classes
    AnyRef
    Annotations
    @throws()
  70. final def wait(arg0: Long): Unit

    Definition Classes
    AnyRef
    Annotations
    @throws()
  71. def zeroMQStream[T](publisherUrl: String, subscribe: Subscribe, bytesToObjects: (Seq[Seq[Byte]]) ⇒ Iterator[T], storageLevel: StorageLevel, supervisorStrategy: SupervisorStrategy)(implicit arg0: ClassManifest[T]): DStream[T]

    Create an input stream that receives messages pushed by a zeromq publisher.

    Create an input stream that receives messages pushed by a zeromq publisher.

    publisherUrl

    Url of remote zeromq publisher

    subscribe

    topic to subscribe to

    bytesToObjects

    A zeroMQ stream publishes sequence of frames for each topic and each frame has sequence of byte thus it needs the converter (which might be deserializer of bytes) to translate from sequence of sequence of bytes, where sequence refer to a frame and sub sequence refer to its payload.

    storageLevel

    RDD storage level. Defaults to memory-only.

Inherited from Logging

Inherited from AnyRef

Inherited from Any