final JavaSparkContext sparkContext = new JavaSparkContext(sparkConf);
JavaStreamingContextFactory streamingContextFactory = new JavaStreamingContextFactory() {
@Override
public JavaStreamingContext create() {
return new JavaStreamingContext(sparkContext, new Duration(batchDurationMS));
}
};
if (checkpointDirString == null) {
log.info("Not using a streaming checkpoint dir");
streamingContext = streamingContextFactory.create();
} else {
log.info("Using streaming checkpoint dir {}", checkpointDirString);
streamingContext = JavaStreamingContext.getOrCreate(
checkpointDirString, sparkContext.hadoopConfiguration(), streamingContextFactory, false);
streamingContext.checkpoint(checkpointDirString);
}
log.info("Creating message queue stream");
JavaPairDStream<K,M> dStream = buildDStream();
if (checkpointDirString != null) {
dStream.checkpoint(new Duration(batchDurationMS));
}
Properties consumerProps = new Properties();
consumerProps.setProperty("group.id", "OryxGroup-SpeedLayer-" + System.currentTimeMillis());
consumerProps.setProperty("zookeeper.connect", updateQueueLockMaster);