State

:: Experimental :: Abstract class for getting and updating the state in mapping function used in the mapWithState operation of a pair DStream (Scala) or a JavaPairDStream (Java).

Scala example of using State:

// A mapping function that maintains an integer state and returns a String
def mappingFunction(key: String, value: Option[Int], state: State[Int]): Option[String] = {
  // Check if state exists
  if (state.exists) {
    val existingState = state.get  // Get the existing state
    val shouldRemove = ...         // Decide whether to remove the state
    if (shouldRemove) {
      state.remove()     // Remove the state
    } else {
      val newState = ...
      state.update(newState)    // Set the new state
    }
  } else {
    val initialState = ...
    state.update(initialState)  // Set the initial state
  }
  ... // return something
}

Java example of using State:

// A mapping function that maintains an integer state and returns a String
Function3<String, Optional<Integer>, State<Integer>, String> mappingFunction =
   new Function3<String, Optional<Integer>, State<Integer>, String>() {

     @Override
     public String call(String key, Optional<Integer> value, State<Integer> state) {
       if (state.exists()) {
         int existingState = state.get(); // Get the existing state
         boolean shouldRemove = ...; // Decide whether to remove the state
         if (shouldRemove) {
           state.remove(); // Remove the state
         } else {
           int newState = ...;
           state.update(newState); // Set the new state
         }
       } else {
         int initialState = ...; // Set the initial state
         state.update(initialState);
       }
       // return something
     }
   };

S: Class of the state

Annotations: @Experimental()
Source: State.scala

Linear Supertypes

AnyRef, Any

Abstract Value Members

abstract def exists(): Boolean

Whether the state already exists
abstract def get(): S

Get the state if it exists, otherwise it will throw java.util.NoSuchElementException.
Get the state if it exists, otherwise it will throw java.util.NoSuchElementException. Check with exists() whether the state exists or not before calling get().

Exceptions thrown
java.util.NoSuchElementException If the state does not exist.
abstract def isTimingOut(): Boolean

Whether the state is timing out and going to be removed by the system after the current batch.
Whether the state is timing out and going to be removed by the system after the current batch. This timeout can occur if timeout duration has been specified in the StatSpec and the key has not received any new data for that timeout duration.
abstract def remove(): Unit

Remove the state if it exists.
Remove the state if it exists.
State cannot be updated if it has been already removed (that is, remove() has already been called) or it is going to be removed due to timeout (that is, isTimingOut() is true).
abstract def update(newState: S): Unit

Update the state with a new value.
Update the state with a new value.
State cannot be updated if it has been already removed (that is, remove() has already been called) or it is going to be removed due to timeout (that is, isTimingOut() is true).

Exceptions thrown
java.lang.IllegalArgumentException If the state has already been removed, or is going to be removed

Concrete Value Members

final def !=(arg0: Any): Boolean

Definition Classes
AnyRef → Any
final def ##(): Int

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

Definition Classes
AnyRef → Any
final def asInstanceOf[T0]: T0

Definition Classes
Any
def clone(): AnyRef

Attributes
protected[java.lang]
Definition Classes
AnyRef
Annotations
@throws( ... )
final def eq(arg0: AnyRef): Boolean

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

Definition Classes
AnyRef → Any
def finalize(): Unit

Attributes
protected[java.lang]
Definition Classes
AnyRef
Annotations
@throws( classOf[java.lang.Throwable] )
final def getClass(): Class[_]

Definition Classes
AnyRef → Any
final def getOption(): Option[S]

Get the state as a scala.Option.
Get the state as a scala.Option. It will be Some(state) if it exists, otherwise None.

Annotations
@inline()
def hashCode(): Int

Definition Classes
AnyRef → Any
final def isInstanceOf[T0]: Boolean

Definition Classes
Any
final def ne(arg0: AnyRef): Boolean

Definition Classes
AnyRef
final def notify(): Unit

Definition Classes
AnyRef
final def notifyAll(): Unit

Definition Classes
AnyRef
final def synchronized[T0](arg0: ⇒ T0): T0

Definition Classes
AnyRef
final def toString(): String

Definition Classes
State → AnyRef → Any
Annotations
@inline()
final def wait(): Unit

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

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

Definition Classes
AnyRef
Annotations
@throws( ... )

Related Doc: package streaming

sealed abstract class State[S] extends AnyRef

Abstract Value Members

abstract def exists(): Boolean

abstract def get(): S

abstract def isTimingOut(): Boolean

abstract def remove(): Unit

abstract def update(newState: S): Unit

Concrete Value Members

final def !=(arg0: Any): Boolean

final def ##(): Int

final def ==(arg0: Any): Boolean

final def asInstanceOf[T0]: T0

def clone(): AnyRef

final def eq(arg0: AnyRef): Boolean

def equals(arg0: Any): Boolean

def finalize(): Unit

final def getClass(): Class[_]

final def getOption(): Option[S]

def hashCode(): Int

final def isInstanceOf[T0]: Boolean

final def ne(arg0: AnyRef): Boolean

final def notify(): Unit

final def notifyAll(): Unit

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

final def toString(): String

final def wait(): Unit

final def wait(arg0: Long, arg1: Int): Unit

final def wait(arg0: Long): Unit

Inherited from AnyRef

Inherited from Any

Ungrouped