Spark streaming in java?

[gsi-gilberto]

Could anyone point me in the right direction to use tranquility with BeamRDD in JAVA?

Basically I would like to see a basic working example with default values just as in:

https://github.com/druid-io/tranquility/blob/master/docs/spark.md

But in java.

[gsi-gilberto]

Does it look something like this?

`
class SimpleBeamFactory implements BeamFactory<Map<String, Object>> {

/**
 * 
 */
private static final long serialVersionUID = 1L;

@Override
public Beam<Map<String, Object>> makeBeam() {

	final CuratorFramework curator = CuratorFrameworkFactory.newClient("druid:2181",
			new BoundedExponentialBackoffRetry(100, 3000, 5));
	curator.start();
	
	Timestamper<Map<String, Object>> timestamper = new Timestamper<Map<String,Object>>() {
		
		/**
		 * 
		 */
		private static final long serialVersionUID = 1L;

		@Override
		public DateTime timestamp(Map<String, Object> arg0) {
			return DateTime.now();
		}
	}; 
	
	final TimestampSpec timestampSpec = new TimestampSpec("time", "auto", null);

	final Beam<Map<String, Object>> beam = DruidBeams.builder(timestamper).curator(curator)
			.discoveryPath("/druid/discovery").location(DruidLocation.create("druid/overlord", "foo"))
			.timestampSpec(timestampSpec)
			.tuning(ClusteredBeamTuning.builder().segmentGranularity(Granularity.SECOND)
					.windowPeriod(new Period(100)).partitions(1).replicants(1).build())
			.druidBeamConfig(DruidBeamConfig.builder().indexRetryPeriod(new Period(500)).build()).buildBeam();

	return beam;
}

}

public class DruidTest {

public static void main(String[] args) throws Exception {

	Properties properties = PropertiesLoader
			.read(DruidTest.class.getClassLoader().getResourceAsStream("spark.properties"));

	String sparkMaster = String.valueOf(properties.get("spark.master"));
	String appName = String.valueOf(properties.get("spark.app.name"));
	String executors = String.valueOf(properties.get("spark.executor.instances"));
	String cores = String.valueOf(properties.get("spark.executor.cores"));
	String testInputPath = String.valueOf(properties.get("test.input.path"));
	Integer windowSeconds = Integer.valueOf(String.valueOf(properties.get("spark.window.seconds")));

	SparkConf sparkConf = new SparkConf().setMaster(sparkMaster).setAppName(appName)
			.set("spark.executor.instances", executors).set("spark.executor.cores", cores);

	@SuppressWarnings("resource")
	JavaStreamingContext jsc = new JavaStreamingContext(sparkConf, Durations.seconds(windowSeconds));

	JavaDStream<String> lines = jsc.textFileStream(testInputPath);

	JavaDStream<String> words = lines
			.flatMap((FlatMapFunction<String, String>) x -> Arrays.asList(x.split(" ")).iterator());

	JavaPairDStream<String, Integer> pairs = words
			.mapToPair((PairFunction<String, String, Integer>) s -> new Tuple2<>(s, 1));

	JavaPairDStream<String, Integer> wordCounts = pairs
			.reduceByKey((Function2<Integer, Integer, Integer>) (i1, i2) -> i1 + i2);

	ClassTag<Tuple2<String, Integer>> tupleClassTag = scala.reflect.ClassTag$.MODULE$.apply(Tuple2.class);

	JavaDStream<Tuple2<String, Integer>> unPaired = new JavaDStream<Tuple2<String, Integer>>(wordCounts.dstream(), tupleClassTag);

	JavaDStream<Map<String,Object>> restructured = unPaired.map(x -> new HashMap<String, Object>() {
		/**
		 * 
		 */
		private static final long serialVersionUID = 1L;

		{
			put(x._1, (Object)x._2);
		}
	});
	
	ClassTag<Map<String, Object>> mapClassTag = scala.reflect.ClassTag$.MODULE$.apply(Map.class);

	restructured.foreachRDD(rdd -> BeamRDD.createBeamRDD(rdd.rdd(), mapClassTag).propagate(new SimpleBeamFactory()));

	jsc.start();
	jsc.awaitTermination();
}

}`