0
votes

I want to combine Measurements by type and device name.

public class Measurement implements Serializable {
private String measurementType;
private String device;
private Long ts;
private Double reading;
}

I am doing an average with Combine.perKey(...) auccessfully by the type. But I want to have basically an compound key of device and measurementType.

Right now my KvDoFn looks like this:

public class KvByMeasurementType extends DoFn<Measurement, KV<String, Measurement>> implements Serializable {
@DoFn.ProcessElement
public void processElement(DoFn<Measurement, KV<String, Measurement>>.ProcessContext context) {
    Measurement measurement = context.element();
    context.output(KV.of(measurement.getMeasurementType(), measurement));
}
}

How to extend it to create a compound key of two values?