org.cripac.isee.vpe.common

Class SparkStreamingApp

java.lang.Object

org.cripac.isee.vpe.common.SparkStreamingApp

All Implemented Interfaces:

Serializable

Direct Known Subclasses:

DataManagingApp, MessageHandlingApp, PedestrianAttrRecogApp, PedestrianReIDUsingAttrApp, PedestrianTrackingApp

public abstract class SparkStreamingApp
extends Object
implements Serializable

The SparkStreamingApp class wraps a whole Spark Streaming application, including driver code and executor code. After initialized, it can be used just like a JavaStreamingContext class. Note that you must call the initialize() method after construction and before using it.

Author:

Ken Yu, CRIPAC, 2016

See Also:

Serialized Form

Field Summary

Fields

Modifier and Type	Field and Description
private String	appName
private org.apache.spark.streaming.api.java.JavaStreamingContext	jssc Common Spark Streaming context variable.
private Map<String,Object>	kafkaParams Kafka parameters for creating input streams pulling messages from Kafka Brokers.
protected Singleton<Logger>	loggerSingleton
private Singleton<MonitorThread>	monitorSingleton
private SystemPropertyCenter	propCenter
private static long	serialVersionUID
private List<Stream>	streams
private Singleton<TaskController>	taskController

Constructor Summary

Constructors

Constructor and Description
SparkStreamingApp(SystemPropertyCenter propCenter, String appName)

Method Summary

Modifier and Type	Method and Description
abstract void	addToContext() Add streaming actions directly to the global streaming context.
void	awaitTermination() Await termination of the application.
protected org.apache.spark.streaming.api.java.JavaPairDStream<DataType,scala.Tuple2<String,byte[]>>	buildDirectStream(Collection<DataType> acceptingTypes) Utility function for all applications to receive messages with byte array values from Kafka with direct stream.
protected org.apache.spark.streaming.api.java.JavaPairDStream<DataType,scala.Tuple2<String,byte[]>>	buildDirectStream(Collection<DataType> acceptingTypes, int repartition) Utility function for all applications to receive messages with byte array values from Kafka with direct stream.
protected void	finalize()
void	initialize() Initialize the application.
protected void	registerStreams(Collection<Stream> streams)
void	start() Start the application.
void	stop() Stop the application.

Methods inherited from class java.lang.Object

clone, equals, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Field Detail

serialVersionUID

private static final long serialVersionUID

See Also:

Constant Field Values

propCenter

@Nonnull
private SystemPropertyCenter propCenter

appName

@Nonnull
private final String appName

kafkaParams

@Nonnull
private final Map<String,Object> kafkaParams

Kafka parameters for creating input streams pulling messages from Kafka Brokers.

jssc

private transient org.apache.spark.streaming.api.java.JavaStreamingContext jssc

Common Spark Streaming context variable.

loggerSingleton

@Nonnull
protected final Singleton<Logger> loggerSingleton

streams

@Nonnull
private final List<Stream> streams

monitorSingleton

@Nonnull
private final Singleton<MonitorThread> monitorSingleton

taskController

@Nullable
private Singleton<TaskController> taskController

Constructor Detail

SparkStreamingApp

public SparkStreamingApp(@Nonnull
                         SystemPropertyCenter propCenter,
                         @Nonnull
                         String appName)
                  throws Exception

Throws:

Exception

Method Detail

registerStreams

protected void registerStreams(Collection<Stream> streams)

buildDirectStream

@Nonnull
protected org.apache.spark.streaming.api.java.JavaPairDStream<DataType,scala.Tuple2<String,byte[]>> buildDirectStream(@Nonnull
                                                                                                                               Collection<DataType> acceptingTypes,
                                                                                                                               int repartition)
                                                                                                                        throws org.apache.spark.SparkException

Utility function for all applications to receive messages with byte array values from Kafka with direct stream.

Parameters:

acceptingTypes - Data types the stream accepts.

repartition - Number of partitions when repartitioning the RDDs. -1 means do not do repartition. 0 means using default parallelism of Spark.

Returns:

A Kafka non-receiver input stream.

Throws:

org.apache.spark.SparkException

buildDirectStream

@Nonnull
protected org.apache.spark.streaming.api.java.JavaPairDStream<DataType,scala.Tuple2<String,byte[]>> buildDirectStream(@Nonnull
                                                                                                                               Collection<DataType> acceptingTypes)
                                                                                                                        throws org.apache.spark.SparkException

Utility function for all applications to receive messages with byte array values from Kafka with direct stream. RDDs are repartitioned by default.

Parameters:

acceptingTypes - Data types the stream accepts.

Returns:

A Kafka non-receiver input stream.

Throws:

org.apache.spark.SparkException

addToContext

public abstract void addToContext()
                           throws Exception

Add streaming actions directly to the global streaming context. This is for applications that may take as input messages with types other than TaskData. Actions that take TaskData as input should be implemented in the Stream.addToGlobalStream(Map), in order to save time of deserialization. Note that existence of Kafka topics used in this method is not automatically checked.

Throws:

Exception

initialize

public void initialize()

Initialize the application.

start

public void start()

Start the application.

stop

public void stop()

Stop the application.

awaitTermination

public void awaitTermination()

Await termination of the application.

finalize

protected void finalize()
                 throws Throwable

Overrides:

finalize in class Object

Throws:

Throwable