PoC Sketch of an DSL for XD.

xd-dsl.groovy

class Event{
	long ts;
	int h;
	int hh;
	int p;
	int property;
	double value;
}

class LoadEvent extends Event{
	//...
}

class WorkEvent extends Event{
	//...
}

def eventForType(Event event){
	switch(event.property){
		case 0 : return new WorkEvent(event)
		case 1 : return new LoadEvent(event)
		default: return event	
	}
}

Composite SmartGrid{
	
	Stream smartGridIngestion(Event event) {
	   
	   http(port=9000)
	   transform(eventForType)
	   filter(expression=event instanceof LoadEvent)
	   hdfs()
	}

	Stream loadStream(LoadEvent event){
	   
	   aggregateCounter(name="smartgrid_{e.h}_{e.hh}_{e.p}_load_actual", timeStamp=event.ts, aggregate=event.value/60)
	}

	Tap loadPrediction(on=smartGridIngestion.filter){

	   predict(model=loadPrediction.pmml, output=(event["load_predicted"]=prediction["predicted_load"]; event))
	   
	   aggregateCounter(name="smartgrid_{e.h}_{e.hh}_{e.p}_load_predicted", timeStamp=event.ts, aggregate=event.value/60)
	}
}