Question - Resample

[sumeshmr]

Hi,
I was trying to use the resample functionality. Its being implemented in Scala, but is not added to Java API(JavaTimeSeriesRDD). How to add this?

[matthjes]

Hi,
You can use TimeSeriesRDD instead of JavaTimeSeriesRDD. Here is an excerpt from the class I'm using to resample power data (in watts) into 5 minute and 1 hour consumption intervals (in watt hours):

public class AggregatorTask {

	private static final Logger LOG = LoggerFactory.getLogger(AggregatorTask.class);

	public static void main(String[] args) throws IOException {
		...
		// create context
		SparkConf conf = new SparkConf().setAppName("Aggregate Time Data");
		JavaSparkContext context = new JavaSparkContext(conf);
		SQLContext sqlContext = new SQLContext(context);

		ZoneId zoneId = ZoneId.systemDefault();
		Dataset<Row> df = ... // load data

		// create DateTimeIndex (timestamps for data)
		// assume 1s interval
		ZonedDateTime start = ZonedDateTime.ofInstant(Instant.ofEpochMilli(1496268001000L), zoneId);
		ZonedDateTime end = ZonedDateTime.ofInstant(Instant.ofEpochMilli(1496354399000L), zoneId);
		DateTimeIndex dtIndex = DateTimeIndexFactory.uniformFromInterval(start, end, new SecondFrequency(1));

		// create time series
		TimeSeriesRDD<String> meterDataRdd = TimeSeriesRDD.timeSeriesRDDFromObservations(dtIndex, df, "timestamp",
				"symbol", "value");

		// cache data in memory
		meterDataRdd.cache();

		// fill gaps using linear interpolation
		meterDataRdd = meterDataRdd.fill("linear");

		// resample data
		resampleAndSave(meterDataRdd, start, end, new MinuteFrequency(5), outputDir, pathSuffix + "\\5m");
		resampleAndSave(meterDataRdd, start, end, new HourFrequency(1), outputDir, pathSuffix + "\\1h");
	}

	private static void resampleAndSave(TimeSeriesRDD<String> meterDataRdd, ZonedDateTime start,
		ZonedDateTime end, Frequency frequency, String outputDirectory, String subDirectory) {
		// resample data to given interval
		DateTimeIndex targetIndex = DateTimeIndexFactory.uniformFromInterval(start, end, frequency);
		TimeSeriesRDD<String> resampled = meterDataRdd.resample(targetIndex, new PowerResampler(),
				false, false);

		// save data to file
		resampled.saveAsCsv(outputDirectory + "\\" + subDirectory);
	}

public class PowerResampler implements Function3<double[], Object, Object, Object>, Serializable {

	@Override
	public Object apply(double[] values, Object startIndex, Object endIndex) {
		// perform numerical integration
		int start = (int) startIndex;
		int end = (int) endIndex;
		double result = 0;
		for (int i = start; i < end - 1; i++) {
			double a = values[i];
			double c = values[i + 1];
			double h = 1.0 / 3600.0;
			result += (a + c) / 2.0 * h;
		}
		return result;
	}

	@Override
	public Function1<double[], Function1<Object, Function1<Object, Object>>> curried() {
		return null;
	}

	@Override
	public Function1<Tuple3<double[], Object, Object>, Object> tupled() {
		return null;
	}
}