Process WindowedKey streams

Author: blootsvoets

src/main/java/org/radarcns/Frequency.java

@@ -17,6 +17,7 @@
 package org.radarcns;
 
 import java.nio.file.Files;
+import java.util.Date;
 import java.util.List;
 import java.util.Objects;
 import javax.annotation.Nonnull;
@@ -63,8 +64,8 @@ public static Frequency read(File file) {
         return new Frequency(file, map);
     }
 
-    public void add(String topicName, String id, GenericRecord valueField, Field timeField) {
-        String timestamp = RestructureAvroRecords.createHourTimestamp(valueField, timeField);
+    public void add(String topicName, String id, Date date) {
+        String timestamp = RestructureAvroRecords.createHourTimestamp(date);
 
         Integer count = (Integer) bins.get(topicName, id, timestamp);
         if (count == null) {

src/main/java/org/radarcns/RestructureAvroRecords.java

@@ -16,17 +16,6 @@
 
 package org.radarcns;
 
-import java.io.File;
-import java.io.FileWriter;
-import java.io.IOException;
-import java.nio.file.Files;
-import java.text.SimpleDateFormat;
-import java.util.ArrayList;
-import java.util.Date;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.TimeZone;
 import org.apache.avro.Schema.Field;
 import org.apache.avro.file.DataFileReader;
 import org.apache.avro.generic.GenericDatumReader;
@@ -45,7 +34,16 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import static java.nio.file.StandardCopyOption.REPLACE_EXISTING;
+import java.io.File;
+import java.io.FileWriter;
+import java.io.IOException;
+import java.text.SimpleDateFormat;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.TimeZone;
 
 public class RestructureAvroRecords {
     private static final Logger logger = LoggerFactory.getLogger(RestructureAvroRecords.class);
@@ -255,8 +253,8 @@ private void writeRecord(GenericRecord record, String topicName, FileCacheStore
             throw new IOException("Failed to process " + record + "; no key or value");
         }
 
-        Field timeField = valueField.getSchema().getField("time");
-        String outputFileName = createFilename(valueField, timeField);
+        Date time = getDate(keyField, valueField);
+        String outputFileName = createFilename(time);
 
         // Clean user id and create final output pathname
         String userId = keyField.get("userId").toString().replaceAll("[^a-zA-Z0-9_-]+", "");
@@ -275,29 +273,42 @@ private void writeRecord(GenericRecord record, String topicName, FileCacheStore
         }
 
         // Count data (binned and total)
-        bins.add(topicName, keyField.get("sourceId").toString(), valueField, timeField);
+        bins.add(topicName, keyField.get("sourceId").toString(), time);
         processedRecordsCount++;
     }
 
-    private String createFilename(GenericRecord valueField, Field timeField) {
-        if (timeField == null) {
-            logger.warn("Time field of record valueField " + valueField + " is not set");
-            return "unknown." + outputFileExtension;
+    private String createFilename(Date date) {
+        if (date == null) {
+            logger.warn("Time field of record valueField is not set");
+            return "unknown_date." + outputFileExtension;
         }
         // Make a timestamped filename YYYYMMDD_HH00.json
-        String hourlyTimestamp = createHourTimestamp(valueField, timeField);
+        String hourlyTimestamp = createHourTimestamp(date);
         return hourlyTimestamp + "00." + outputFileExtension;
     }
 
-    public static String createHourTimestamp(GenericRecord valueField, Field timeField) {
-        if (timeField == null) {
-            return "unknown";
+    public static String createHourTimestamp(Date date) {
+        if (date == null) {
+            return "unknown_date";
         }
 
-        double time = (Double) valueField.get(timeField.pos());
-        // Convert from millis to date and apply dateFormat
-        Date date = new Date((long) (time * 1000d));
         return FILE_DATE_FORMAT.format(date);
     }
 
+    public static Date getDate(GenericRecord keyField, GenericRecord valueField) {
+        Field timeField = valueField.getSchema().getField("time");
+        if (timeField != null) {
+            double time = (Double) valueField.get(timeField.pos());
+            // Convert from millis to date and apply dateFormat
+            return new Date((long) (time * 1000d));
+        }
+
+        // WindowedKey
+        timeField = keyField.getSchema().getField("start");
+        if (timeField == null) {
+            return null;
+        }
+        long time = (Long) keyField.get("start");
+        return new Date(time);
+    }
 }

src/test/java/org/radarcns/RestructureAvroRecordsTest.java

@@ -33,15 +33,32 @@ public class RestructureAvroRecordsTest {
     @Test
     public void createHourTimestamp() throws Exception {
         long currentTime = 1493711175;  // Tue May  2 07:46:15 UTC 2017
+        long startTime = (currentTime - 3600) * 1000L;
 
-        Schema schema = SchemaBuilder.record("value").fields()
+        Schema keySchema = SchemaBuilder.record("value").fields()
+                .name("start").type("long").noDefault()
+                .endRecord();
+        GenericRecord keyField = new GenericRecordBuilder(keySchema)
+                .set("start", startTime).build();
+
+        Schema valueSchema1 = SchemaBuilder.record("value").fields()
                 .name("time").type("double").noDefault()
                 .endRecord();
-        GenericRecord record = new GenericRecordBuilder(schema).set("time",
-                (double)currentTime).build();
-        Field timeField = schema.getField("time");
-        String result = RestructureAvroRecords.createHourTimestamp(record, timeField);
+        GenericRecord valueField1 = new GenericRecordBuilder(valueSchema1)
+                .set("time", (double)currentTime).build();
+
+        Date date = RestructureAvroRecords.getDate(keyField, valueField1);
+        String result = RestructureAvroRecords.createHourTimestamp(date);
 
         assertEquals("20170502_07", result);
+
+        Schema valueSchema2 = SchemaBuilder.record("value").fields()
+                .name("a").type("double").noDefault()
+                .endRecord();
+        GenericRecord valueField2 = new GenericRecordBuilder(valueSchema2)
+                .set("a", 0.1).build();
+        date = RestructureAvroRecords.getDate(keyField, valueField2);
+        result = RestructureAvroRecords.createHourTimestamp(date);
+        assertEquals("20170502_06", result);
     }
 }