pubsub json support + common transform from value map to featurerow

ingestion/src/main/java/feast/options/OptionsParser.java

@@ -22,7 +22,6 @@
 import com.fasterxml.jackson.module.jsonSchema.JsonSchema;
 import com.fasterxml.jackson.module.jsonSchema.JsonSchemaGenerator;
 import com.google.common.collect.Lists;
-import feast.source.csv.CsvFileFeatureSource.CsvFileFeatureSourceOptions;
 import java.io.IOException;
 import java.util.List;
 import java.util.Map;

ingestion/src/main/java/feast/source/csv/CsvIO.java → ingestion/src/main/java/feast/source/common/ParseCsvTransform.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018 The Feast Authors
+ * Copyright 2019 The Feast Authors
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -15,7 +15,7 @@
  *
  */
 
-package feast.source.csv;
+package feast.source.common;
 
 import java.io.IOException;
 import java.io.Serializable;
@@ -26,21 +26,41 @@
 import java.util.stream.Collectors;
 import lombok.Builder;
 import lombok.NonNull;
-import org.apache.beam.sdk.coders.SerializableCoder;
-import org.apache.beam.sdk.io.TextIO;
+import org.apache.beam.sdk.coders.MapCoder;
+import org.apache.beam.sdk.coders.StringUtf8Coder;
 import org.apache.beam.sdk.transforms.DoFn;
 import org.apache.beam.sdk.transforms.PTransform;
 import org.apache.beam.sdk.transforms.ParDo;
 import org.apache.beam.sdk.values.PCollection;
-import org.apache.beam.sdk.values.PInput;
 import org.apache.commons.csv.CSVFormat;
 import org.apache.commons.csv.CSVParser;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
-public class CsvIO {
-
-  private static final Logger LOGGER = LoggerFactory.getLogger(CsvIO.class);
+/**
+ * Transform for reading text CSV jsonfiles into a map of strings to strings. CSV format is assumed
+ * to be RFC4180.
+ */
+@Builder
+public class ParseCsvTransform extends
+    PTransform<PCollection<String>, PCollection<Map<String, String>>> {
+
+  @NonNull
+  private List<String> header;
+
+  @Override
+  public PCollection<Map<String, String>> expand(PCollection<String> input) {
+    CSVLineParser csvLineParser = new CSVLineParser(header);
+    return input.apply(ParDo.of(new DoFn<String, Map<String, String>>() {
+      @ProcessElement
+      public void processElement(ProcessContext context) {
+        String line = context.element();
+        if (line != null && !line.isEmpty()) {
+          for (StringMap map : csvLineParser.records(line)) {
+            context.output(map);
+          }
+        }
+      }
+    })).setCoder(MapCoder.of(StringUtf8Coder.of(), StringUtf8Coder.of()));
+  }
 
   public static class StringMap extends HashMap<String, String> {
 
@@ -59,44 +79,6 @@ public StringMap thisput(String key, String value) {
     }
   }
 
-
-  public static Read read(String inputPath, List<String> header) {
-    return Read.builder()
-        .inputPath(inputPath)
-        .csvLineParser(new CSVLineParser(header))
-        .build();
-  }
-
-  /**
-   * Transform for reading text CSV jsonfiles into a map of strings to strings. CSV format is assumed to
-   * be RFC4180.
-   */
-  @Builder
-  public static class Read extends PTransform<PInput, PCollection<StringMap>> {
-
-    @NonNull
-    private final CSVLineParser csvLineParser;
-    @NonNull
-    private String inputPath;
-
-    @Override
-    public PCollection<StringMap> expand(PInput input) {
-      PCollection<String> text = input.getPipeline().apply(TextIO.read().from(inputPath));
-      return text.apply(ParDo.of(new DoFn<String, StringMap>() {
-        @ProcessElement
-        public void processElement(ProcessContext context) {
-          String line = context.element();
-          if (line != null && !line.isEmpty()) {
-            for (StringMap map : csvLineParser.records(line)) {
-              context.output(map);
-            }
-          }
-        }
-      })).setCoder(SerializableCoder.of(StringMap.class));
-    }
-  }
-
-
   /**
    * This is a helper class to make it easy to use the commons csv parser object for one line
    * without recreating it.
@@ -123,5 +105,4 @@ public List<StringMap> records(String input) {
       }
     }
   }
-
 }

ingestion/src/main/java/feast/source/common/ParseJsonTransform.java

@@ -0,0 +1,110 @@
+/*
+ * Copyright 2019 The Feast Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package feast.source.common;
+
+import com.google.gson.Gson;
+import com.google.gson.GsonBuilder;
+import com.google.gson.JsonDeserializationContext;
+import com.google.gson.JsonDeserializer;
+import com.google.gson.JsonElement;
+import com.google.gson.JsonParseException;
+import com.google.gson.JsonPrimitive;
+import com.google.gson.reflect.TypeToken;
+import feast.ingestion.model.Values;
+import feast.types.ValueProto.Value;
+import java.lang.reflect.Type;
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.util.Map;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.values.PCollection;
+
+public class ParseJsonTransform extends
+    PTransform<PCollection<String>, PCollection<Map<String, Value>>> {
+
+  @Override
+  public PCollection<Map<String, Value>> expand(PCollection<String> input) {
+    return null;
+  }
+
+  private static class ValueDeserializer implements JsonDeserializer<Value> {
+
+    @Override
+    public Value deserialize(JsonElement json, Type typeOfT, JsonDeserializationContext context)
+        throws JsonParseException {
+      if (json.isJsonArray() || json.isJsonObject()) {
+        throw new JsonParseException("Expecting primitive value");
+      }
+      if (json.isJsonNull()) {
+        return Value.newBuilder().build(); // return UNKNOWN value.
+      }
+      if (json.isJsonPrimitive()) {
+        JsonPrimitive primitive = json.getAsJsonPrimitive();
+        if (primitive.isBoolean()) {
+          return Values.ofBool(primitive.getAsBoolean());
+        } else if (primitive.isString()) {
+          return Values.ofString(primitive.getAsString());
+        } else {
+          BigDecimal bigDec = json.getAsBigDecimal();
+          // Find out if it is an int type
+          try {
+            BigInteger bigInt = bigDec.toBigIntegerExact();
+            try {
+              return Values.ofInt32(bigInt.intValueExact());
+            } catch (ArithmeticException e) {
+            }
+            return Values.ofInt64(bigDec.longValue());
+          } catch (ArithmeticException e) {
+          }
+          try {
+            return Values.ofFloat(bigDec.floatValue());
+          } catch (ArithmeticException e) {
+          }
+          // Just return it as a double
+          return Values.ofDouble(bigDec.doubleValue());
+        }
+        return null;
+      }
+    }
+
+    public static class ParseJsonDoFn extends DoFn<String, Map<String, Value>> {
+
+      private transient Gson gson;
+      private transient Type valueMapType;
+
+
+      @ProcessElement
+      public void processElement(ProcessContext context) {
+        context.output(parseJson(context.element()));
+      }
+
+      Map<String, Value> parseJson(String json) {
+        if (gson == null) {
+          GsonBuilder gsonBuilder = new GsonBuilder();
+          gsonBuilder.registerTypeAdapter(Object.class, new ValueDeserializer());
+          gson = gsonBuilder.create();
+          valueMapType = new TypeToken<Map<String, Value>>() {
+          }.getType();
+        }
+        return gson.fromJson(json, valueMapType);
+      }
+    }
+
+  }
+}

ingestion/src/main/java/feast/source/common/StringToValueMapTransform.java

@@ -0,0 +1,60 @@
+/*
+ * Copyright 2019 The Feast Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package feast.source.common;
+
+import com.google.common.base.Strings;
+import feast.ingestion.model.Values;
+import feast.types.ValueProto.Value;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Map.Entry;
+import org.apache.beam.sdk.coders.MapCoder;
+import org.apache.beam.sdk.coders.StringUtf8Coder;
+import org.apache.beam.sdk.extensions.protobuf.ProtoCoder;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.values.PCollection;
+
+public class StringToValueMapTransform extends
+    PTransform<PCollection<Map<String, String>>, PCollection<Map<String, Value>>> {
+
+
+  @Override
+  public PCollection<Map<String, Value>> expand(PCollection<Map<String, String>> input) {
+    return input.apply(ParDo.of(new StringToValueMapDoFn()))
+        .setCoder(MapCoder.of(StringUtf8Coder.of(),
+            ProtoCoder.of(Value.class)));
+  }
+
+  public static class StringToValueMapDoFn extends
+      DoFn<Map<String, String>, Map<String, Value>> {
+
+    @ProcessElement
+    public void processElement(ProcessContext context) {
+      Map<String, Value> row = new HashMap<>();
+      for (Entry<String, String> entry : context.element().entrySet()) {
+        String stringVal = entry.getValue();
+        if (!Strings.isNullOrEmpty(stringVal)) {
+          row.put(entry.getKey(), Values.ofString(entry.getValue()));
+        }
+      }
+      context.output(row);
+    }
+  }
+}

ingestion/src/main/java/feast/source/common/ValueMap.java

@@ -0,0 +1,25 @@
+/*
+ * Copyright 2019 The Feast Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package feast.source.common;
+
+import feast.types.ValueProto.Value;
+import java.util.HashMap;
+
+public class ValueMap extends HashMap<String, Value> {
+
+}