Repository: gojek/feast
Pull request #94 (2019-01-19 03:50:16)

Allow submission of kafka jobs
Fix for the submission validation bug mentioned in #22.

feast-ci-bot: [APPROVALNOTIFIER] This PR is **NOT APPROVED**

This pull-request has been approved by:
To fully approve this pull request, please assign additional approvers.
We suggest the following additional approver: **zhilingc**

If they are not already assigned, you can assign the PR to them by writing `/assign @zhilingc` in a comment when ready.

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands).

The pull request process is described [here](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process)

<details open>
Needs approval from an approver in each of these files:

- **[OWNERS](https://github.com/gojek/feast/blob/master/OWNERS)**

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":["zhilingc"]} -->

Patch:
diff --git a/core/src/main/java/feast/core/validators/SpecValidator.java b/core/src/main/java/feast/core/validators/SpecValidator.java
index 643374626..c8500ac33 100644
--- a/core/src/main/java/feast/core/validators/SpecValidator.java
+++ b/core/src/main/java/feast/core/validators/SpecValidator.java
@@ -17,6 +17,10 @@
 
 package feast.core.validators;
 
+import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Preconditions.checkNotNull;
+import static feast.core.validators.Matchers.checkLowerSnakeCase;
+
 import com.google.common.base.Strings;
 import com.google.common.collect.Lists;
 import feast.core.dao.EntityInfoRepository;
@@ -35,36 +39,28 @@ import feast.specs.FeatureSpecProto.FeatureSpec;
 import feast.specs.ImportSpecProto.Field;
 import feast.specs.ImportSpecProto.ImportSpec;
 import feast.specs.StorageSpecProto.StorageSpec;
-import org.springframework.beans.factory.annotation.Autowired;
-
 import java.util.Arrays;
 import java.util.Optional;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
-
-import static com.google.common.base.Preconditions.checkArgument;
-import static com.google.common.base.Preconditions.checkNotNull;
-import static feast.core.validators.Matchers.checkLowerSnakeCase;
+import org.springframework.beans.factory.annotation.Autowired;
 
 public class SpecValidator {
 
-  private StorageInfoRepository storageInfoRepository;
-  private EntityInfoRepository entityInfoRepository;
-  private FeatureGroupInfoRepository featureGroupInfoRepository;
-  private FeatureInfoRepository featureInfoRepository;
   private static final String FILE_ERROR_STORE_TYPE = "file.json";
-
   private static final String NO_STORE = "";
-
   private static String[] SUPPORTED_WAREHOUSE_STORES =
-      new String[] {
+      new String[]{
           BigQueryStorageManager.TYPE, FILE_ERROR_STORE_TYPE,
       };
-
   private static String[] SUPPORTED_SERVING_STORES =
-      new String[] {
+      new String[]{
           BigTableStorageManager.TYPE, PostgresStorageManager.TYPE, RedisStorageManager.TYPE,
       };
+  private StorageInfoRepository storageInfoRepository;
+  private EntityInfoRepository entityInfoRepository;
+  private FeatureGroupInfoRepository featureGroupInfoRepository;
+  private FeatureInfoRepository featureInfoRepository;
 
   @Autowired
   public SpecValidator(
@@ -130,7 +126,8 @@ public class SpecValidator {
         servingStoreId =
             servingStoreId.equals(NO_STORE) ? group.getServingStore().getId() : servingStoreId;
         warehouseStoreId =
-            warehouseStoreId.equals(NO_STORE) ? group.getWarehouseStore().getId() : warehouseStoreId;
+            warehouseStoreId.equals(NO_STORE) ? group.getWarehouseStore().getId()
+                : warehouseStoreId;
       }
       Optional<StorageInfo> servingStore = storageInfoRepository.findById(servingStoreId);
       Optional<StorageInfo> warehouseStore = storageInfoRepository.findById(warehouseStoreId);
@@ -221,10 +218,14 @@ public class SpecValidator {
   public void validateImportSpec(ImportSpec spec) throws IllegalArgumentException {
     try {
       switch (spec.getType()) {
+        case "kafka":
+          checkKafkaImportSpecOption(spec);
+          break;
         case "pubsub":
           checkPubSubImportSpecOption(spec);
           break;
-        case "file":
+        case "file.csv":
+        case "file.json":
           checkFileImportSpecOption(spec);
           checkArgument(
               !spec.getSchema().getEntityIdColumn().equals(""),
@@ -261,11 +262,22 @@ public class SpecValidator {
     }
   }
 
+  private void checkKafkaImportSpecOption(ImportSpec spec) {
+    try {
+      String topics = spec.getOptionsOrDefault("topics", "");
+      String server = spec.getOptionsOrDefault("server", "");
+      if (topics.equals("") && server.equals("")) {
+        throw new IllegalArgumentException(
+            "Kafka ingestion requires either topics or servers");
+      }
+    } catch (NullPointerException | IllegalArgumentException e) {
+      throw new IllegalArgumentException(
+          Strings.lenientFormat("Invalid options: %s", e.getMessage()));
+    }
+  }
+
   private void checkFileImportSpecOption(ImportSpec spec) throws IllegalArgumentException {
     try {
-      checkArgument(
-          Lists.newArrayList("json", "csv").contains(spec.getOptionsOrThrow("format")),
-          "File format must be of type 'json' or 'csv'");
       checkArgument(!spec.getOptionsOrDefault("path", "").equals(""), "File path cannot be empty");
     } catch (NullPointerException | IllegalArgumentException e) {
       throw new IllegalArgumentException(
diff --git a/ingestion/src/main/java/feast/ingestion/transform/FeatureEnums.java b/ingestion/src/main/java/feast/ingestion/transform/FeatureEnums.java
deleted file mode 100644
index 77985082c..000000000
--- a/ingestion/src/main/java/feast/ingestion/transform/FeatureEnums.java
+++ /dev/null
@@ -1,15 +0,0 @@
-package feast.ingestion.transform;
-
-public class FeatureEnums {
-    public enum InputSource {
-        FILE,
-        BIGQUERY,
-        PUBSUB,
-        KAFKA
-    }
-
-    public enum FileFormat {
-        CSV,
-        JSON
-    }
-}
\ No newline at end of file
diff --git a/ingestion/src/main/java/feast/ingestion/transform/FeatureRowCsvIO.java b/ingestion/src/main/java/feast/ingestion/transform/FeatureRowCsvIO.java
deleted file mode 100644
index fb1bfddb3..000000000
--- a/ingestion/src/main/java/feast/ingestion/transform/FeatureRowCsvIO.java
+++ /dev/null
@@ -1,135 +0,0 @@
-/*
- * Copyright 2018 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.ingestion.transform;
-
-import com.google.common.base.Preconditions;
-import com.google.common.base.Strings;
-import com.google.common.collect.Lists;
-import com.google.common.collect.Maps;
-import feast.ingestion.metrics.FeastMetrics;
-import feast.ingestion.model.Values;
-import feast.ingestion.transform.CsvIO.StringMap;
-import feast.ingestion.util.DateUtil;
-import feast.specs.ImportSpecProto.Field;
-import feast.specs.ImportSpecProto.ImportSpec;
-import feast.types.FeatureProto.Feature;
-import feast.types.FeatureRowProto.FeatureRow;
-import lombok.Builder;
-import lombok.NonNull;
-import org.apache.beam.sdk.transforms.DoFn;
-import org.apache.beam.sdk.transforms.ParDo;
-import org.apache.beam.sdk.values.PCollection;
-import org.apache.beam.sdk.values.PInput;
-
-import java.util.List;
-import java.util.Map;
-import java.util.Map.Entry;
-
-public class FeatureRowCsvIO {
-
-  public static Read read(ImportSpec importSpec) {
-    return Read.builder().importSpec(importSpec).build();
-  }
-
-  /**
-   * Transform for processing CSV text jsonfiles and producing FeatureRow messages. CSV jsonfiles
-   * with headers are not supported.
-   *
-   * <p>This transform asserts that the import spec is for only one entity, as all columns must have
-   * the same entity. There must be the same number of columns in the CSV jsonfiles as in the import
-   * spec.
-   *
-   * <p>The output is a PCollection of {@link feast.types.FeatureRowProto.FeatureRow FeatureRows},
-   * where every feature and entity {@link feast.types.ValueProto.Value Value} in the FeatureRow is
-   * a string (set via Value.stringVal).
-   */
-  @Builder
-  public static class Read extends FeatureIO.Read {
-
-    @NonNull private final ImportSpec importSpec;
-
-    @Override
-    public PCollection<FeatureRow> expand(PInput input) {
-      final List<String> fieldNames = Lists.newArrayList();
-      final Map<String, Field> fields = Maps.newHashMap();
-      for (Field field : importSpec.getSchema().getFieldsList()) {
-        String displayName = !field.getName().isEmpty() ? field.getName() : field.getFeatureId();
-        fieldNames.add(displayName);
-        fields.put(displayName, field);
-      }
-
-      String path = importSpec.getOptionsMap().get("path");
-      Preconditions.checkArgument(
-          !Strings.isNullOrEmpty(path), "path must be set as option in ImportSpec for CSV import");
-      Preconditions.checkArgument(
-          importSpec.getEntitiesList().size() == 1, "exactly 1 entity must be set for CSV import");
-      String entity = importSpec.getEntities(0);
-      String entityIdColumn = importSpec.getSchema().getEntityIdColumn();
-      Preconditions.checkArgument(
-          !Strings.isNullOrEmpty(entityIdColumn), "entity id column must be set");
-      String timestampColumn = importSpec.getSchema().getTimestampColumn();
-      Preconditions.checkArgument(
-          importSpec.getSchema().getFieldsList().size() > 0,
-          "CSV import needs schema with a least one field specified");
-
-      PCollection<StringMap> stringMaps = input.getPipeline().apply(CsvIO.read(path, fieldNames));
-
-      return stringMaps.apply(
-          ParDo.of(
-              new DoFn<StringMap, FeatureRow>() {
-                @ProcessElement
-                public void processElement(ProcessContext context) {
-                  FeatureRow.Builder builder = FeatureRow.newBuilder();
-                  try {
-                    StringMap stringMap = context.element();
-                    builder.setEntityName(entity);
-
-
-                    for (Entry<String, String> entry : stringMap.entrySet()) {
-                      String name = entry.getKey();
-                      String value = entry.getValue();
-                      Field field = fields.get(name);
-
-                      // A feature can only be one of these things
-                      if (entityIdColumn.equals(name)) {
-                        builder.setEntityKey(value);
-                      } else if (timestampColumn.equals(name)) {
-                        builder.setEventTimestamp(DateUtil.toTimestamp(value));
-                      } else if (!Strings.isNullOrEmpty(field.getFeatureId())) {
-                        String featureId = field.getFeatureId();
-                        builder.addFeatures(
-                            Feature.newBuilder().setId(featureId).setValue(Values.ofString(value)));
-                      }
-                      // else silently ignore this column
-                    }
-                    if (!importSpec
-                        .getSchema()
-                        .getTimestampValue()
-                        .equals(com.google.protobuf.Timestamp.getDefaultInstance())) {
-                      // This overrides any column event timestamp.
-                      builder.setEventTimestamp(importSpec.getSchema().getTimestampValue());
-                    }
-                    context.output(builder.build());
-                  } catch (Exception e) {
-                    FeastMetrics.inc(builder.build(), "input_errors");
-                  }
-                }
-              }));
-    }
-  }
-}
diff --git a/ingestion/src/main/java/feast/ingestion/transform/FeatureRowJsonTextIO.java b/ingestion/src/main/java/feast/ingestion/transform/FeatureRowJsonTextIO.java
deleted file mode 100644
index 5814334d8..000000000
--- a/ingestion/src/main/java/feast/ingestion/transform/FeatureRowJsonTextIO.java
+++ /dev/null
@@ -1,78 +0,0 @@
-/*
- * Copyright 2018 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.ingestion.transform;
-
-import com.google.common.base.Preconditions;
-import com.google.protobuf.InvalidProtocolBufferException;
-import com.google.protobuf.util.JsonFormat;
-import feast.specs.ImportSpecProto.ImportSpec;
-import feast.types.FeatureRowProto.FeatureRow;
-import lombok.Builder;
-import lombok.NonNull;
-import org.apache.beam.sdk.io.TextIO;
-import org.apache.beam.sdk.transforms.DoFn;
-import org.apache.beam.sdk.transforms.ParDo;
-import org.apache.beam.sdk.values.PCollection;
-import org.apache.beam.sdk.values.PInput;
-
-public class FeatureRowJsonTextIO {
-
-  public static Read read(ImportSpec importSpec) {
-    return Read.builder().importSpec(importSpec).build();
-  }
-
-  /**
-   * Transform for processing JSON text jsonfiles and that contain JSON serialised FeatureRow messages,
-   * one per line.
-   *
-   * <p>This transform asserts that the import spec is for only one entity, as all columns must have
-   * the same entity.
-   *
-   * <p>The output is a PCollection of {@link feast.types.FeatureRowProto.FeatureRow FeatureRows}.
-   */
-  @Builder
-  public static class Read extends FeatureIO.Read {
-
-    @NonNull private final ImportSpec importSpec;
-
-    @Override
-    public PCollection<FeatureRow> expand(PInput input) {
-      String path = importSpec.getOptionsOrDefault("path", null);
-      Preconditions.checkNotNull(path, "Path must be set in for file import");
-      PCollection<String> jsonLines = input.getPipeline().apply(TextIO.read().from(path));
-
-      return jsonLines.apply(
-          ParDo.of(
-              new DoFn<String, FeatureRow>() {
-                @ProcessElement
-                public void processElement(ProcessContext context) {
-                  String line = context.element();
-                  FeatureRow.Builder builder = FeatureRow.newBuilder();
-                  try {
-                    // TODO use proto registry so that it can read Any feature values, not just
-                    // primitives.
-                    JsonFormat.parser().merge(line, builder);
-                    context.output(builder.build());
-                  } catch (InvalidProtocolBufferException e) {
-                    throw new RuntimeException(e);
-                  }
-                }
-              }));
-    }
-  }
-}
diff --git a/ingestion/src/main/java/feast/ingestion/transform/FeatureRowKafkaIO.java b/ingestion/src/main/java/feast/ingestion/transform/FeatureRowKafkaIO.java
deleted file mode 100644
index eeb282759..000000000
--- a/ingestion/src/main/java/feast/ingestion/transform/FeatureRowKafkaIO.java
+++ /dev/null
@@ -1,97 +0,0 @@
-/*
- * Copyright 2018 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.ingestion.transform;
-
-import static com.google.common.base.Preconditions.checkArgument;
-
-import feast.ingestion.deserializer.FeatureRowDeserializer;
-import feast.ingestion.deserializer.FeatureRowKeyDeserializer;
-import feast.options.Options;
-import feast.options.OptionsParser;
-import feast.specs.ImportSpecProto.ImportSpec;
-import feast.types.FeatureRowProto.FeatureRow;
-import feast.types.FeatureRowProto.FeatureRowKey;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.List;
-import javax.validation.constraints.NotEmpty;
-import org.apache.beam.sdk.io.kafka.KafkaIO;
-import org.apache.beam.sdk.io.kafka.KafkaRecord;
-import org.apache.beam.sdk.transforms.DoFn;
-import org.apache.beam.sdk.transforms.ParDo;
-import org.apache.beam.sdk.values.PCollection;
-import org.apache.beam.sdk.values.PInput;
-
-public class FeatureRowKafkaIO {
-  static final String KAFKA_TYPE = "kafka";
-
-  /**
-   * Transform for reading {@link feast.types.FeatureRowProto.FeatureRow FeatureRow} proto messages
-   * from kafka one or more kafka topics.
-   */
-  public static Read read(ImportSpec importSpec) {
-    return new Read(importSpec);
-  }
-
-  public static class KafkaReadOptions implements Options {
-    @NotEmpty public String server;
-    @NotEmpty public String topics;
-  }
-
-  public static class Read extends FeatureIO.Read {
-
-    private ImportSpec importSpec;
-
-    private Read(ImportSpec importSpec) {
-      this.importSpec = importSpec;
-    }
-
-    @Override
-    public PCollection<FeatureRow> expand(PInput input) {
-
-      checkArgument(importSpec.getType().equals(KAFKA_TYPE));
-
-      KafkaReadOptions options =
-          OptionsParser.parse(importSpec.getOptionsMap(), KafkaReadOptions.class);
-
-      List<String> topicsList = new ArrayList<>(Arrays.asList(options.topics.split(",")));
-
-      KafkaIO.Read<FeatureRowKey, FeatureRow> kafkaIOReader =
-          KafkaIO.<FeatureRowKey, FeatureRow>read()
-              .withBootstrapServers(options.server)
-              .withTopics(topicsList)
-              .withKeyDeserializer(FeatureRowKeyDeserializer.class)
-              .withValueDeserializer(FeatureRowDeserializer.class);
-
-      PCollection<KafkaRecord<FeatureRowKey, FeatureRow>> featureRowRecord =
-          input.getPipeline().apply(kafkaIOReader);
-
-      PCollection<FeatureRow> featureRow =
-          featureRowRecord.apply(
-              ParDo.of(
-                  new DoFn<KafkaRecord<FeatureRowKey, FeatureRow>, FeatureRow>() {
-                    @ProcessElement
-                    public void processElement(ProcessContext processContext) {
-                      KafkaRecord<FeatureRowKey, FeatureRow> record = processContext.element();
-                      processContext.output(record.getKV().getValue());
-                    }
-                  }));
-      return featureRow;
-    }
-  }
-}
diff --git a/ingestion/src/main/java/feast/ingestion/transform/FeatureRowPubSubIO.java b/ingestion/src/main/java/feast/ingestion/transform/FeatureRowPubSubIO.java
deleted file mode 100644
index bb0d2c335..000000000
--- a/ingestion/src/main/java/feast/ingestion/transform/FeatureRowPubSubIO.java
+++ /dev/null
@@ -1,87 +0,0 @@
-/*
- * Copyright 2018 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.ingestion.transform;
-
-import static com.google.common.base.Preconditions.checkArgument;
-
-import javax.validation.constraints.AssertTrue;
-import org.apache.beam.sdk.io.gcp.pubsub.PubsubIO;
-import org.apache.beam.sdk.values.PCollection;
-import org.apache.beam.sdk.values.PInput;
-import feast.options.Options;
-import feast.options.OptionsParser;
-import feast.specs.ImportSpecProto.ImportSpec;
-import feast.types.FeatureRowProto.FeatureRow;
-
-public class FeatureRowPubSubIO {
-
-  public static final String PUB_SUB_TYPE = "pubsub";
-
-  public static Read read(ImportSpec importSpec) {
-    return new Read(importSpec);
-  }
-
-  /**
-   * Transform for reading {@link feast.types.FeatureRowProto.FeatureRow FeatureRow}
-   * proto messages from Cloud PubSub.
-   *
-   * <p>This transform accepts multiple entities in the import spec and expects no columns to be
-   * specified as it does not need to construct FeatureRows, merely pass them on.
-   *
-   * <p>Because Feast ingestion is stateless, the message event time is simply the processing time,
-   * there is no need to override it based on any property of the message.
-   */
-  public static class Read extends FeatureIO.Read {
-
-    private ImportSpec importSpec;
-
-    private Read(ImportSpec importSpec) {
-      this.importSpec = importSpec;
-    }
-
-    @Override
-    public PCollection<FeatureRow> expand(PInput input) {
-      checkArgument(importSpec.getType().equals(PUB_SUB_TYPE));
-      PubSubReadOptions options =
-          OptionsParser.parse(importSpec.getOptionsMap(), PubSubReadOptions.class);
-
-      PubsubIO.Read<FeatureRow> read = readProtos();
-
-      if (options.subscription != null) {
-        read = read.fromSubscription(options.subscription);
-      } else if (options.topic != null) {
-        read = read.fromTopic(options.topic);
-      }
-      return input.getPipeline().apply(read);
-    }
-
-    PubsubIO.Read<FeatureRow> readProtos() {
-      return PubsubIO.readProtos(FeatureRow.class);
-    }
-
-    public static class PubSubReadOptions implements Options {
-      public String subscription;
-      public String topic;
-
-      @AssertTrue(message = "subscription or topic must be set")
-      boolean isValid() {
-        return subscription != null || topic != null;
-      }
-    }
-  }
-}
diff --git a/ingestion/src/main/java/feast/ingestion/transform/ReadFeaturesTransform.java b/ingestion/src/main/java/feast/ingestion/transform/ReadFeaturesTransform.java
index af7945bc7..5da8f6e75 100644
--- a/ingestion/src/main/java/feast/ingestion/transform/ReadFeaturesTransform.java
+++ b/ingestion/src/main/java/feast/ingestion/transform/ReadFeaturesTransform.java
@@ -17,10 +17,10 @@
 
 package feast.ingestion.transform;
 
-import com.google.common.base.Enums;
 import com.google.common.base.Preconditions;
-import feast.storage.bigquery.FeatureRowBigQueryIO;
 import com.google.inject.Inject;
+import feast.source.FeatureSourceFactory;
+import feast.source.FeatureSourceFactoryService;
 import feast.specs.ImportSpecProto.ImportSpec;
 import feast.types.FeatureRowProto.FeatureRow;
 import org.apache.beam.sdk.transforms.PTransform;
@@ -29,43 +29,31 @@ import org.apache.beam.sdk.values.PInput;
 
 public class ReadFeaturesTransform extends PTransform<PInput, PCollection<FeatureRow>> {
 
-    private ImportSpec importSpec;
+  private ImportSpec importSpec;
 
-    @Inject
-    public ReadFeaturesTransform(ImportSpec importSpec) {
-        this.importSpec = importSpec;
-    }
+  @Inject
+  public ReadFeaturesTransform(ImportSpec importSpec) {
+    this.importSpec = importSpec;
+  }
 
-    @Override
-    public PCollection<FeatureRow> expand(PInput input) {
-        return input.getPipeline().apply("Read " + importSpec.getType(), getTransform());
-    }
+  @Override
+  public PCollection<FeatureRow> expand(PInput input) {
+    return input.getPipeline().apply("Read " + importSpec.getType(), getTransform());
+  }
+
+  public PTransform<PInput, PCollection<FeatureRow>> getTransform() {
+    String type = importSpec.getType();
+    Preconditions.checkArgument(!type.isEmpty(), "type missing in import spec");
 
-    public PTransform<PInput, PCollection<FeatureRow>> getTransform() {
-        String type = importSpec.getType();
-        Preconditions.checkArgument(!type.isEmpty(), "type missing in import spec");
-        Preconditions.checkArgument(Enums.getIfPresent(FeatureEnums.InputSource.class, type.toUpperCase()).isPresent(), "The type defined is invalid or not supported");
-        switch (FeatureEnums.InputSource.valueOf(type.toUpperCase())) {
-            case FILE:
-                String format = importSpec.getOptionsOrDefault("format", null);
-                Preconditions.checkNotNull(format, "format option missing from import spec of type file");
-                switch (FeatureEnums.FileFormat.valueOf(format.toUpperCase())) {
-                    case CSV:
-                        return FeatureRowCsvIO.read(importSpec);
-                    case JSON:
-                        return FeatureRowJsonTextIO.read(importSpec);
-                    default:
-                        throw new IllegalArgumentException("Unknown format in import spec" + type);
-                }
-            case BIGQUERY:
-                return FeatureRowBigQueryIO.read(importSpec);
-            case PUBSUB:
-                return FeatureRowPubSubIO.read(importSpec);
-            case KAFKA:
-                return FeatureRowKafkaIO.read(importSpec);
-            default:
-                throw new IllegalArgumentException("Unknown type in import spec" + type);
-        }
+    FeatureSourceFactory featureSourceFactory = null;
+    for (FeatureSourceFactory factory : FeatureSourceFactoryService.getAll()) {
+      if (type.equals(factory.getType())) {
+        featureSourceFactory = factory;
+      }
     }
+    Preconditions
+        .checkNotNull(featureSourceFactory, "No FeatureSourceFactory found for type " + type);
+    return featureSourceFactory.create(importSpec);
+  }
 }
 
diff --git a/ingestion/src/main/java/feast/ingestion/util/JsonUtil.java b/ingestion/src/main/java/feast/ingestion/util/JsonUtil.java
index b142c013f..d32a79d1c 100644
--- a/ingestion/src/main/java/feast/ingestion/util/JsonUtil.java
+++ b/ingestion/src/main/java/feast/ingestion/util/JsonUtil.java
@@ -1,3 +1,20 @@
+/*
+ * 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
+ *
+ * http://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.ingestion.util;
 
 import com.google.gson.Gson;
diff --git a/ingestion/src/main/java/feast/options/Options.java b/ingestion/src/main/java/feast/options/Options.java
index 920674bb2..47077810c 100644
--- a/ingestion/src/main/java/feast/options/Options.java
+++ b/ingestion/src/main/java/feast/options/Options.java
@@ -19,5 +19,9 @@ package feast.options;
 
 import java.io.Serializable;
 
-/** iterface for identifying classes that can use the OptionsParser for extra type safety */
-public interface Options extends Serializable {}
+/**
+ * interface for identifying classes that can use the OptionsParser for extra type safety
+ */
+public interface Options extends Serializable {
+
+}
diff --git a/ingestion/src/main/java/feast/options/OptionsParser.java b/ingestion/src/main/java/feast/options/OptionsParser.java
index 33001ca06..7eab26963 100644
--- a/ingestion/src/main/java/feast/options/OptionsParser.java
+++ b/ingestion/src/main/java/feast/options/OptionsParser.java
@@ -22,6 +22,7 @@ import com.fasterxml.jackson.databind.ObjectMapper;
 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;
@@ -32,6 +33,7 @@ import javax.validation.Validator;
 import javax.validation.ValidatorFactory;
 
 public class OptionsParser {
+
   private static final ObjectMapper mapper = new ObjectMapper();
   private static final Validator validator;
 
@@ -41,7 +43,9 @@ public class OptionsParser {
     }
   }
 
-  /** Return a json schema string representing an options class for error messages */
+  /**
+   * Return a json schema string representing an options class for error messages
+   */
   static <T extends Options> String getJsonSchema(Class<T> optionsClass) {
     JsonSchemaGenerator schemaGen = new JsonSchemaGenerator(mapper);
     JsonSchema schema = null;
@@ -54,7 +58,9 @@ public class OptionsParser {
     }
   }
 
-  /** Construct a class from string options and validate with any javax validation annotations */
+  /**
+   * Construct a class from string options and validate with any javax validation annotations
+   */
   public static <T extends Options> T parse(Map<String, String> optionsMap, Class<T> clazz) {
     List<String> messages = Lists.newArrayList();
     T options;
diff --git a/ingestion/src/main/java/feast/storage/bigquery/BigQuerySourceOptions.java b/ingestion/src/main/java/feast/source/FeatureSource.java
similarity index 57%
rename from ingestion/src/main/java/feast/storage/bigquery/BigQuerySourceOptions.java
rename to ingestion/src/main/java/feast/source/FeatureSource.java
index aff8cded6..e4c0590b3 100644
--- a/ingestion/src/main/java/feast/storage/bigquery/BigQuerySourceOptions.java
+++ b/ingestion/src/main/java/feast/source/FeatureSource.java
@@ -1,11 +1,11 @@
 /*
- * 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.
  * You may obtain a copy of the License at
  *
- *     https://www.apache.org/licenses/LICENSE-2.0
+ * http://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,
@@ -15,13 +15,13 @@
  *
  */
 
-package feast.storage.bigquery;
+package feast.source;
 
-import javax.validation.constraints.NotEmpty;
-import feast.options.Options;
+import feast.types.FeatureRowProto.FeatureRow;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.PInput;
+
+public abstract class FeatureSource extends PTransform<PInput, PCollection<FeatureRow>> {
 
-public class BigQuerySourceOptions implements Options {
-  @NotEmpty public String project;
-  @NotEmpty public String dataset;
-  @NotEmpty public String table;
 }
diff --git a/ingestion/src/main/java/feast/storage/FeatureSource.java b/ingestion/src/main/java/feast/source/FeatureSourceFactory.java
similarity index 64%
rename from ingestion/src/main/java/feast/storage/FeatureSource.java
rename to ingestion/src/main/java/feast/source/FeatureSourceFactory.java
index 78f5db558..fc4d64418 100644
--- a/ingestion/src/main/java/feast/storage/FeatureSource.java
+++ b/ingestion/src/main/java/feast/source/FeatureSourceFactory.java
@@ -1,11 +1,11 @@
 /*
- * 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.
  * You may obtain a copy of the License at
  *
- *     https://www.apache.org/licenses/LICENSE-2.0
+ * http://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,
@@ -15,13 +15,17 @@
  *
  */
 
-package feast.storage;
+package feast.source;
 
-import feast.ingestion.transform.FeatureIO;
 import feast.specs.ImportSpecProto.ImportSpec;
 
-public interface FeatureSource {
-  FeatureIO.Read create(ImportSpec importSpec);
+/**
+ * A FeatureSourceFactory creates FeatureSource instances, which can read FeatureRow messages from a
+ * source location.
+ */
+public interface FeatureSourceFactory {
 
   String getType();
+
+  FeatureSource create(ImportSpec importSpec);
 }
diff --git a/ingestion/src/main/java/feast/source/FeatureSourceFactoryService.java b/ingestion/src/main/java/feast/source/FeatureSourceFactoryService.java
new file mode 100644
index 000000000..18917cb1d
--- /dev/null
+++ b/ingestion/src/main/java/feast/source/FeatureSourceFactoryService.java
@@ -0,0 +1,61 @@
+/*
+ * 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
+ *
+ * http://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;
+
+import avro.shaded.com.google.common.collect.Lists;
+import com.google.common.collect.Iterators;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.ServiceLoader;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class FeatureSourceFactoryService {
+
+  private static ServiceLoader<FeatureSourceFactory> serviceLoader = ServiceLoader
+      .load(FeatureSourceFactory.class);
+  private static List<FeatureSourceFactory> manuallyRegistered = new ArrayList<>();
+
+  static {
+    for (FeatureSourceFactory source : getAll()) {
+      log.info("FeatureSourceFactory type found: " + source.getType());
+    }
+  }
+
+  public static List<FeatureSourceFactory> getAll() {
+    return Lists.newArrayList(
+        Iterators.concat(manuallyRegistered.iterator(), serviceLoader.iterator()));
+  }
+
+  /**
+   * Get store of the given subclass.
+   */
+  public static <T extends FeatureSourceFactory> T get(Class<T> clazz) {
+    for (FeatureSourceFactory store : getAll()) {
+      if (clazz.isInstance(store)) {
+        //noinspection unchecked
+        return (T) store;
+      }
+    }
+    return null;
+  }
+
+  public static void register(FeatureSourceFactory store) {
+    manuallyRegistered.add(store);
+  }
+}
diff --git a/ingestion/src/main/java/feast/source/bigquery/BigQueryFeatureSource.java b/ingestion/src/main/java/feast/source/bigquery/BigQueryFeatureSource.java
new file mode 100644
index 000000000..9ce7f965f
--- /dev/null
+++ b/ingestion/src/main/java/feast/source/bigquery/BigQueryFeatureSource.java
@@ -0,0 +1,108 @@
+/*
+ * 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
+ *
+ * http://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.bigquery;
+
+import static com.google.common.base.Preconditions.checkArgument;
+
+import com.google.auto.service.AutoService;
+import com.google.common.base.Preconditions;
+import feast.options.Options;
+import feast.options.OptionsParser;
+import feast.source.FeatureSource;
+import feast.source.FeatureSourceFactory;
+import feast.specs.ImportSpecProto.ImportSpec;
+import feast.types.FeatureRowProto.FeatureRow;
+import java.util.List;
+import javax.validation.constraints.NotEmpty;
+import lombok.AllArgsConstructor;
+import lombok.NonNull;
+import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.PInput;
+
+
+/**
+ * Transform for processing BigQuery tables and producing FeatureRow messages.
+ *
+ * <p>This transform asserts that the import spec is for only one entity, as all columns must have
+ * the same entity. The columns names in the import spec will be used for selecting columns from the
+ * BigQuery table, but it still scans the whole row.
+ *
+ * <p>The output is a PCollection of {@link feast.types.FeatureRowProto.FeatureRow FeatureRows},
+ * where each feature and the entity key {@link feast.types.ValueProto.Value Value} in the
+ * FeatureRow is taken from a column in the BigQuery table and set with the closest type to the
+ * BigQuery column schema that is available in {@link feast.types.ValueProto.ValueType ValueType}.
+ *
+ * <p>Note a small gotcha is that because Integers and Numerics in BigQuery are 64 bits, these are
+ * always cast to INT64 and DOUBLE respectively.
+ *
+ * <p>Downstream these will fail validation if the corresponding {@link
+ * feast.specs.FeatureSpecProto.FeatureSpec FeatureSpec} has a 32 bit type.
+ */
+@AllArgsConstructor
+public class BigQueryFeatureSource extends FeatureSource {
+
+  private static final String BIGQUERY_FEATURE_SOURCE_TYPE = "bigquery";
+
+  @NonNull
+  private final ImportSpec importSpec;
+
+  @Override
+  public PCollection<FeatureRow> expand(PInput input) {
+    BigQuerySourceOptions options = OptionsParser
+        .parse(importSpec.getOptionsMap(), BigQuerySourceOptions.class);
+
+    List<String> entities = importSpec.getEntitiesList();
+    Preconditions.checkArgument(
+        entities.size() == 1, "exactly 1 entity must be set for BigQuery import");
+
+    String url = String.format("%s:%s.%s", options.project, options.dataset, options.table);
+
+    return input
+        .getPipeline()
+        .apply(
+            BigQueryIO.read(new BigQueryToFeatureRowFn(importSpec)).from(url));
+  }
+
+
+  public static class BigQuerySourceOptions implements Options {
+
+    @NotEmpty
+    public String project;
+    @NotEmpty
+    public String dataset;
+    @NotEmpty
+    public String table;
+  }
+
+
+  @AutoService(FeatureSourceFactory.class)
+  public static class Factory implements FeatureSourceFactory {
+
+    @Override
+    public String getType() {
+      return BIGQUERY_FEATURE_SOURCE_TYPE;
+    }
+
+    @Override
+    public FeatureSource create(ImportSpec importSpec) {
+      checkArgument(importSpec.getType().equals(getType()));
+      return new BigQueryFeatureSource(importSpec);
+    }
+  }
+}
diff --git a/ingestion/src/main/java/feast/storage/bigquery/FeatureRowFromBigQuerySchemaAndRecordFn.java b/ingestion/src/main/java/feast/source/bigquery/BigQueryToFeatureRowFn.java
similarity index 92%
rename from ingestion/src/main/java/feast/storage/bigquery/FeatureRowFromBigQuerySchemaAndRecordFn.java
rename to ingestion/src/main/java/feast/source/bigquery/BigQueryToFeatureRowFn.java
index 243c496cc..795735410 100644
--- a/ingestion/src/main/java/feast/storage/bigquery/FeatureRowFromBigQuerySchemaAndRecordFn.java
+++ b/ingestion/src/main/java/feast/source/bigquery/BigQueryToFeatureRowFn.java
@@ -15,7 +15,7 @@
  *
  */
 
-package feast.storage.bigquery;
+package feast.source.bigquery;
 
 import com.google.api.services.bigquery.model.TableFieldSchema;
 import com.google.api.services.bigquery.model.TableSchema;
@@ -23,6 +23,7 @@ import com.google.cloud.bigquery.LegacySQLTypeName;
 import com.google.cloud.bigquery.StandardSQLTypeName;
 import com.google.common.collect.Maps;
 import com.google.protobuf.Timestamp;
+import feast.storage.bigquery.ValueBigQueryBuilder;
 import java.util.Map;
 import org.apache.avro.generic.GenericRecord;
 import org.apache.beam.sdk.io.gcp.bigquery.SchemaAndRecord;
@@ -40,16 +41,13 @@ import org.slf4j.LoggerFactory;
  * This is a serializable function used with the BigQueryIO for fetching feature rows directly from
  * BigQuery
  */
-public class FeatureRowFromBigQuerySchemaAndRecordFn
+public class BigQueryToFeatureRowFn
     implements SerializableFunction<SchemaAndRecord, FeatureRow> {
 
-  private static final Logger LOGGER =
-      LoggerFactory.getLogger(FeatureRowFromBigQuerySchemaAndRecordFn.class);
-
   private final ImportSpec importSpec;
   private final Map<String, Field> fields;
 
-  public FeatureRowFromBigQuerySchemaAndRecordFn(ImportSpec importSpec) {
+  public BigQueryToFeatureRowFn(ImportSpec importSpec) {
     this.importSpec = importSpec;
     fields = Maps.newHashMap();
     for (Field field : importSpec.getSchema().getFieldsList()) {
diff --git a/ingestion/src/main/java/feast/source/csv/CsvFileFeatureSource.java b/ingestion/src/main/java/feast/source/csv/CsvFileFeatureSource.java
new file mode 100644
index 000000000..837050fc6
--- /dev/null
+++ b/ingestion/src/main/java/feast/source/csv/CsvFileFeatureSource.java
@@ -0,0 +1,161 @@
+/*
+ * Copyright 2018 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.csv;
+
+import static com.google.common.base.Preconditions.checkArgument;
+
+import com.google.auto.service.AutoService;
+import com.google.common.base.Preconditions;
+import com.google.common.base.Strings;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import feast.ingestion.metrics.FeastMetrics;
+import feast.ingestion.model.Values;
+import feast.ingestion.util.DateUtil;
+import feast.options.Options;
+import feast.options.OptionsParser;
+import feast.source.FeatureSource;
+import feast.source.FeatureSourceFactory;
+import feast.source.csv.CsvIO.StringMap;
+import feast.specs.ImportSpecProto.Field;
+import feast.specs.ImportSpecProto.ImportSpec;
+import feast.specs.ImportSpecProto.Schema;
+import feast.types.FeatureProto.Feature;
+import feast.types.FeatureRowProto.FeatureRow;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import javax.validation.constraints.NotEmpty;
+import lombok.AllArgsConstructor;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.PInput;
+
+
+/**
+ * Transform for processing CSV text jsonfiles and producing FeatureRow messages. CSV jsonfiles with
+ * headers are not supported.
+ *
+ * <p>This transform asserts that the import spec is for only one entity, as all columns must have
+ * the same entity. There must be the same number of columns in the CSV jsonfiles as in the import
+ * spec.
+ *
+ * <p>The output is a PCollection of {@link feast.types.FeatureRowProto.FeatureRow FeatureRows},
+ * where every feature and entity {@link feast.types.ValueProto.Value Value} in the FeatureRow is a
+ * string (set via Value.stringVal).
+ */
+@AllArgsConstructor
+public class CsvFileFeatureSource extends FeatureSource {
+
+  public static final String CSV_FILE_FEATURE_SOURCE_TYPE = "file.csv";
+
+  private final ImportSpec importSpec;
+
+  @Override
+  public PCollection<FeatureRow> expand(PInput input) {
+    CsvFileFeatureSourceOptions options = OptionsParser
+        .parse(importSpec.getOptionsMap(), CsvFileFeatureSourceOptions.class);
+    List<String> entities = importSpec.getEntitiesList();
+    Preconditions.checkArgument(
+        entities.size() == 1, "exactly 1 entity must be set for CSV import");
+    Schema schema = importSpec.getSchema();
+    String entity = entities.get(0);
+
+    final List<String> fieldNames = Lists.newArrayList();
+    final Map<String, Field> fields = Maps.newHashMap();
+    for (Field field : schema.getFieldsList()) {
+      String displayName = !field.getName().isEmpty() ? field.getName() : field.getFeatureId();
+      fieldNames.add(displayName);
+      fields.put(displayName, field);
+    }
+
+    String path = options.path;
+    String entityIdColumn = schema.getEntityIdColumn();
+    Preconditions.checkArgument(
+        !Strings.isNullOrEmpty(entityIdColumn), "entity id column must be set");
+    String timestampColumn = schema.getTimestampColumn();
+    Preconditions.checkArgument(
+        schema.getFieldsList().size() > 0,
+        "CSV import needs schema with a least one field specified");
+
+    PCollection<StringMap> stringMaps = input.getPipeline().apply(CsvIO.read(path, fieldNames));
+
+    return stringMaps.apply(
+        ParDo.of(
+            new DoFn<StringMap, FeatureRow>() {
+              @ProcessElement
+              public void processElement(ProcessContext context) {
+                FeatureRow.Builder builder = FeatureRow.newBuilder();
+                try {
+                  StringMap stringMap = context.element();
+                  builder.setEntityName(entity);
+
+                  for (Entry<String, String> entry : stringMap.entrySet()) {
+                    String name = entry.getKey();
+                    String value = entry.getValue();
+                    Field field = fields.get(name);
+
+                    // A feature can only be one of these things
+                    if (entityIdColumn.equals(name)) {
+                      builder.setEntityKey(value);
+                    } else if (timestampColumn.equals(name)) {
+                      builder.setEventTimestamp(DateUtil.toTimestamp(value));
+                    } else if (!Strings.isNullOrEmpty(field.getFeatureId())) {
+                      String featureId = field.getFeatureId();
+                      builder.addFeatures(
+                          Feature.newBuilder().setId(featureId).setValue(Values.ofString(value)));
+                    }
+                    // else silently ignore this column
+                  }
+                  if (!schema.getTimestampValue()
+                      .equals(com.google.protobuf.Timestamp.getDefaultInstance())) {
+                    // This overrides any column event timestamp.
+                    builder.setEventTimestamp(schema.getTimestampValue());
+                  }
+                  context.output(builder.build());
+                } catch (Exception e) {
+                  FeastMetrics.inc(builder.build(), "input_errors");
+                }
+              }
+            }));
+
+  }
+
+
+  public static class CsvFileFeatureSourceOptions implements Options {
+
+    @NotEmpty
+    public String path;
+  }
+
+  @AutoService(FeatureSourceFactory.class)
+  public static class Factory implements FeatureSourceFactory {
+
+    @Override
+    public String getType() {
+      return CSV_FILE_FEATURE_SOURCE_TYPE;
+    }
+
+    @Override
+    public FeatureSource create(ImportSpec importSpec) {
+      checkArgument(importSpec.getType().equals(getType()));
+      return new CsvFileFeatureSource(importSpec);
+    }
+  }
+}
diff --git a/ingestion/src/main/java/feast/ingestion/transform/CsvIO.java b/ingestion/src/main/java/feast/source/csv/CsvIO.java
similarity index 99%
rename from ingestion/src/main/java/feast/ingestion/transform/CsvIO.java
rename to ingestion/src/main/java/feast/source/csv/CsvIO.java
index 4e8cec196..e6cf0f9dc 100644
--- a/ingestion/src/main/java/feast/ingestion/transform/CsvIO.java
+++ b/ingestion/src/main/java/feast/source/csv/CsvIO.java
@@ -15,7 +15,7 @@
  *
  */
 
-package feast.ingestion.transform;
+package feast.source.csv;
 
 import java.io.IOException;
 import java.io.Serializable;
diff --git a/ingestion/src/main/java/feast/source/json/JsonFileFeatureSource.java b/ingestion/src/main/java/feast/source/json/JsonFileFeatureSource.java
new file mode 100644
index 000000000..a38a2e0b8
--- /dev/null
+++ b/ingestion/src/main/java/feast/source/json/JsonFileFeatureSource.java
@@ -0,0 +1,101 @@
+/*
+ * Copyright 2018 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.json;
+
+import static com.google.common.base.Preconditions.checkArgument;
+
+import com.google.auto.service.AutoService;
+import com.google.protobuf.InvalidProtocolBufferException;
+import com.google.protobuf.util.JsonFormat;
+import feast.options.Options;
+import feast.options.OptionsParser;
+import feast.source.FeatureSource;
+import feast.source.FeatureSourceFactory;
+import feast.specs.ImportSpecProto.ImportSpec;
+import feast.types.FeatureRowProto.FeatureRow;
+import javax.validation.constraints.NotEmpty;
+import lombok.AllArgsConstructor;
+import org.apache.beam.sdk.io.TextIO;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.PInput;
+
+
+/**
+ * Transform for processing JSON text jsonfiles and that contain JSON serialised FeatureRow
+ * messages, one per line.
+ *
+ * <p>This transform asserts that the import spec is for only one entity, as all columns must
+ * have the same entity.
+ *
+ * <p>The output is a PCollection of {@link feast.types.FeatureRowProto.FeatureRow FeatureRows}.
+ */
+@AllArgsConstructor
+public class JsonFileFeatureSource extends FeatureSource {
+
+  public static final String JSON_FILE_FEATURE_SOURCE_TYPE = "file.json";
+
+  private final ImportSpec importSpec;
+
+  @Override
+  public PCollection<FeatureRow> expand(PInput input) {
+    JsonFileFeatureSourceOptions options = OptionsParser
+        .parse(importSpec.getOptionsMap(), JsonFileFeatureSourceOptions.class);
+    PCollection<String> jsonLines = input.getPipeline().apply(TextIO.read().from(options.path));
+    return jsonLines.apply(
+        ParDo.of(
+            new DoFn<String, FeatureRow>() {
+              @ProcessElement
+              public void processElement(ProcessContext context) {
+                String line = context.element();
+                FeatureRow.Builder builder = FeatureRow.newBuilder();
+                try {
+                  JsonFormat.parser().merge(line, builder);
+                  context.output(builder.build());
+                } catch (InvalidProtocolBufferException e) {
+                  throw new RuntimeException(e);
+                }
+              }
+            }));
+  }
+
+  public static class JsonFileFeatureSourceOptions implements Options {
+
+    @NotEmpty
+    public String path;
+  }
+
+  @AutoService(FeatureSourceFactory.class)
+  public static class Factory implements FeatureSourceFactory {
+
+    @Override
+    public String getType() {
+      return JSON_FILE_FEATURE_SOURCE_TYPE;
+    }
+
+    @Override
+    public FeatureSource create(ImportSpec importSpec) {
+      checkArgument(importSpec.getType().equals(getType()));
+
+      return new JsonFileFeatureSource(importSpec);
+    }
+  }
+
+}
+
diff --git a/ingestion/src/main/java/feast/source/kafka/KafkaFeatureSource.java b/ingestion/src/main/java/feast/source/kafka/KafkaFeatureSource.java
new file mode 100644
index 000000000..c16308871
--- /dev/null
+++ b/ingestion/src/main/java/feast/source/kafka/KafkaFeatureSource.java
@@ -0,0 +1,108 @@
+/*
+ * Copyright 2018 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.kafka;
+
+import static com.google.common.base.Preconditions.checkArgument;
+
+import com.google.auto.service.AutoService;
+import feast.options.Options;
+import feast.options.OptionsParser;
+import feast.source.FeatureSource;
+import feast.source.FeatureSourceFactory;
+import feast.source.kafka.deserializer.FeatureRowDeserializer;
+import feast.source.kafka.deserializer.FeatureRowKeyDeserializer;
+import feast.specs.ImportSpecProto.ImportSpec;
+import feast.types.FeatureRowProto.FeatureRow;
+import feast.types.FeatureRowProto.FeatureRowKey;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import javax.validation.constraints.NotEmpty;
+import lombok.AllArgsConstructor;
+import org.apache.beam.sdk.io.kafka.KafkaIO;
+import org.apache.beam.sdk.io.kafka.KafkaRecord;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.PInput;
+
+/**
+ * Transform for reading {@link feast.types.FeatureRowProto.FeatureRow FeatureRow} proto messages
+ * from kafka one or more kafka topics.
+ */
+@AllArgsConstructor
+public class KafkaFeatureSource extends FeatureSource {
+
+  public static final String KAFKA_FEATURE_SOURCE_TYPE = "kafka";
+  private ImportSpec importSpec;
+
+
+  @Override
+  public PCollection<FeatureRow> expand(PInput input) {
+    checkArgument(importSpec.getType().equals(KAFKA_FEATURE_SOURCE_TYPE));
+
+    KafkaReadOptions options =
+        OptionsParser.parse(importSpec.getOptionsMap(), KafkaReadOptions.class);
+
+    List<String> topicsList = new ArrayList<>(Arrays.asList(options.topics.split(",")));
+
+    KafkaIO.Read<FeatureRowKey, FeatureRow> kafkaIOReader =
+        KafkaIO.<FeatureRowKey, FeatureRow>read()
+            .withBootstrapServers(options.server)
+            .withTopics(topicsList)
+            .withKeyDeserializer(FeatureRowKeyDeserializer.class)
+            .withValueDeserializer(FeatureRowDeserializer.class);
+
+    PCollection<KafkaRecord<FeatureRowKey, FeatureRow>> featureRowRecord =
+        input.getPipeline().apply(kafkaIOReader);
+
+    return
+        featureRowRecord.apply(
+            ParDo.of(
+                new DoFn<KafkaRecord<FeatureRowKey, FeatureRow>, FeatureRow>() {
+                  @ProcessElement
+                  public void processElement(ProcessContext processContext) {
+                    KafkaRecord<FeatureRowKey, FeatureRow> record = processContext.element();
+                    processContext.output(record.getKV().getValue());
+                  }
+                }));
+  }
+
+  public static class KafkaReadOptions implements Options {
+
+    @NotEmpty
+    public String server;
+    @NotEmpty
+    public String topics;
+  }
+
+  @AutoService(FeatureSourceFactory.class)
+  public static class Factory implements FeatureSourceFactory {
+
+    @Override
+    public String getType() {
+      return KAFKA_FEATURE_SOURCE_TYPE;
+    }
+
+    @Override
+    public FeatureSource create(ImportSpec importSpec) {
+      checkArgument(importSpec.getType().equals(getType()));
+      return new KafkaFeatureSource(importSpec);
+    }
+  }
+}
diff --git a/ingestion/src/main/java/feast/ingestion/deserializer/FeatureRowDeserializer.java b/ingestion/src/main/java/feast/source/kafka/deserializer/FeatureRowDeserializer.java
similarity index 97%
rename from ingestion/src/main/java/feast/ingestion/deserializer/FeatureRowDeserializer.java
rename to ingestion/src/main/java/feast/source/kafka/deserializer/FeatureRowDeserializer.java
index c6ab40831..e75a17e64 100644
--- a/ingestion/src/main/java/feast/ingestion/deserializer/FeatureRowDeserializer.java
+++ b/ingestion/src/main/java/feast/source/kafka/deserializer/FeatureRowDeserializer.java
@@ -15,7 +15,7 @@
  *
  */
 
-package feast.ingestion.deserializer;
+package feast.source.kafka.deserializer;
 
 import com.google.protobuf.InvalidProtocolBufferException;
 import feast.types.FeatureRowProto.FeatureRow;
diff --git a/ingestion/src/main/java/feast/ingestion/deserializer/FeatureRowKeyDeserializer.java b/ingestion/src/main/java/feast/source/kafka/deserializer/FeatureRowKeyDeserializer.java
similarity index 97%
rename from ingestion/src/main/java/feast/ingestion/deserializer/FeatureRowKeyDeserializer.java
rename to ingestion/src/main/java/feast/source/kafka/deserializer/FeatureRowKeyDeserializer.java
index 57574d698..4e77d7c99 100644
--- a/ingestion/src/main/java/feast/ingestion/deserializer/FeatureRowKeyDeserializer.java
+++ b/ingestion/src/main/java/feast/source/kafka/deserializer/FeatureRowKeyDeserializer.java
@@ -15,7 +15,7 @@
  *
  */
 
-package feast.ingestion.deserializer;
+package feast.source.kafka.deserializer;
 
 import com.google.protobuf.InvalidProtocolBufferException;
 import feast.types.FeatureRowProto.FeatureRowKey;
diff --git a/ingestion/src/main/java/feast/source/pubsub/PubSubFeatureSource.java b/ingestion/src/main/java/feast/source/pubsub/PubSubFeatureSource.java
new file mode 100644
index 000000000..209ba2b5e
--- /dev/null
+++ b/ingestion/src/main/java/feast/source/pubsub/PubSubFeatureSource.java
@@ -0,0 +1,101 @@
+/*
+ * Copyright 2018 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.pubsub;
+
+import static com.google.common.base.Preconditions.checkArgument;
+
+import com.google.auto.service.AutoService;
+import feast.options.Options;
+import feast.options.OptionsParser;
+import feast.source.FeatureSource;
+import feast.source.FeatureSourceFactory;
+import feast.specs.ImportSpecProto.ImportSpec;
+import feast.types.FeatureRowProto.FeatureRow;
+import javax.validation.constraints.AssertTrue;
+import lombok.Builder;
+import lombok.NonNull;
+import org.apache.beam.sdk.io.gcp.pubsub.PubsubIO;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.PInput;
+import org.apache.hadoop.hbase.util.Strings;
+
+
+/**
+ * Transform for reading {@link feast.types.FeatureRowProto.FeatureRow FeatureRow} proto messages
+ * from Cloud PubSub.
+ *
+ * <p>This transform accepts multiple entities in the import spec and expects no columns to be
+ * specified as it does not need to construct FeatureRows, merely pass them on.
+ *
+ * <p>Because Feast ingestion is stateless, the message event time is simply the processing time,
+ * there is no need to override it based on any property of the message.
+ */
+@Builder
+public class PubSubFeatureSource extends FeatureSource {
+
+  public static final String PUBSUB_FEATURE_SOURCE_TYPE = "pubsub";
+
+  @NonNull
+  private ImportSpec importSpec;
+
+  @Override
+  public PCollection<FeatureRow> expand(PInput input) {
+    checkArgument(importSpec.getType().equals(PUBSUB_FEATURE_SOURCE_TYPE));
+    PubSubReadOptions options =
+        OptionsParser.parse(importSpec.getOptionsMap(), PubSubReadOptions.class);
+
+    PubsubIO.Read<FeatureRow> read = readProtos();
+
+    if (!Strings.isEmpty(options.subscription)) {
+      read = read.fromSubscription(options.subscription);
+    } else if (!Strings.isEmpty(options.topic)) {
+      read = read.fromTopic(options.topic);
+    }
+    return input.getPipeline().apply(read);
+  }
+
+  PubsubIO.Read<FeatureRow> readProtos() {
+    return PubsubIO.readProtos(FeatureRow.class);
+  }
+
+  public static class PubSubReadOptions implements Options {
+
+    public String subscription;
+    public String topic;
+
+    @AssertTrue(message = "subscription or topic must be set")
+    boolean isValid() {
+      return !Strings.isEmpty(subscription) || !Strings.isEmpty(topic);
+    }
+  }
+
+  @AutoService(FeatureSourceFactory.class)
+  public static class Factory implements FeatureSourceFactory {
+
+    @Override
+    public String getType() {
+      return PUBSUB_FEATURE_SOURCE_TYPE;
+    }
+
+    @Override
+    public FeatureSource create(ImportSpec importSpec) {
+      checkArgument(importSpec.getType().equals(getType()));
+      return new PubSubFeatureSource(importSpec);
+    }
+  }
+}
diff --git a/ingestion/src/main/java/feast/storage/bigquery/FeatureRowBigQueryIO.java b/ingestion/src/main/java/feast/storage/bigquery/FeatureRowBigQueryIO.java
index 5384a0744..b5e039cc9 100644
--- a/ingestion/src/main/java/feast/storage/bigquery/FeatureRowBigQueryIO.java
+++ b/ingestion/src/main/java/feast/storage/bigquery/FeatureRowBigQueryIO.java
@@ -18,20 +18,16 @@
 package feast.storage.bigquery;
 
 import com.google.api.services.bigquery.model.TableSchema;
-import com.google.common.base.Preconditions;
 import com.google.common.base.Strings;
 import com.google.inject.Inject;
 import feast.ingestion.model.Specs;
 import feast.ingestion.transform.FeatureIO;
 import feast.ingestion.transform.SplitFeatures.SingleOutputSplit;
-import feast.options.OptionsParser;
 import feast.specs.EntitySpecProto.EntitySpec;
 import feast.specs.FeatureSpecProto.FeatureSpec;
-import feast.specs.ImportSpecProto.ImportSpec;
 import feast.types.FeatureRowExtendedProto.FeatureRowExtended;
 import feast.types.FeatureRowProto.FeatureRow;
 import feast.types.GranularityProto.Granularity;
-import lombok.AllArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO;
 import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write.CreateDisposition;
@@ -43,59 +39,12 @@ import org.apache.beam.sdk.io.gcp.bigquery.WriteResult;
 import org.apache.beam.sdk.options.ValueProvider.StaticValueProvider;
 import org.apache.beam.sdk.values.PCollection;
 import org.apache.beam.sdk.values.PDone;
-import org.apache.beam.sdk.values.PInput;
 import org.apache.beam.sdk.values.ValueInSingleWindow;
 import org.joda.time.Duration;
 
 @Slf4j
 public class FeatureRowBigQueryIO {
 
-  public static Read read(ImportSpec importSpec) {
-    // TODO: Allow and convert from /projects/{project}/datasets/{dataset}/tables/{table}, to
-    // {project}:{dataset}.{table}
-    return new Read(importSpec);
-  }
-
-  /**
-   * Transform for processing BigQuery tables and producing FeatureRow messages.
-   *
-   * <p>This transform asserts that the import spec is for only one entity, as all columns must have
-   * the same entity. The columns names in the import spec will be used for selecting columns from
-   * the BigQuery table, but it still scans the whole row.
-   *
-   * <p>The output is a PCollection of {@link feast.types.FeatureRowProto.FeatureRow FeatureRows},
-   * where each feature and the entity key {@link feast.types.ValueProto.Value Value} in the
-   * FeatureRow is taken from a column in the BigQuery table and set with the closest type to the
-   * BigQuery column schema that is available in {@link feast.types.ValueProto.ValueType ValueType}.
-   *
-   * <p>Note a small gotcha is that because Integers and Numerics in BigQuery are 64 bits, these are
-   * always cast to INT64 and DOUBLE respectively.
-   *
-   * <p>Downstream these will fail validation if the corresponding {@link
-   * feast.specs.FeatureSpecProto.FeatureSpec FeatureSpec} has a 32 bit type.
-   */
-  @AllArgsConstructor
-  public static class Read extends FeatureIO.Read {
-
-    private final ImportSpec importSpec;
-
-    @Override
-    public PCollection<FeatureRow> expand(PInput input) {
-      Preconditions.checkArgument(importSpec.getType().equals("bigquery"));
-      BigQuerySourceOptions options =
-          OptionsParser.parse(importSpec.getOptionsMap(), BigQuerySourceOptions.class);
-
-      String url = String.format("%s:%s.%s", options.project, options.dataset, options.table);
-
-      Preconditions.checkArgument(
-          importSpec.getEntitiesCount() == 1, "BigQuery read must have only one entity");
-      return input
-          .getPipeline()
-          .apply(
-              BigQueryIO.read(new FeatureRowFromBigQuerySchemaAndRecordFn(importSpec)).from(url));
-    }
-  }
-
   public static class Write extends FeatureIO.Write {
 
     private final BigQueryStoreOptions bigQueryOptions;
diff --git a/ingestion/src/main/java/feast/storage/file/json/JsonFileFeatureIO.java b/ingestion/src/main/java/feast/storage/file/json/JsonFileFeatureIO.java
index 0a2f0d32b..0ebbbcb54 100644
--- a/ingestion/src/main/java/feast/storage/file/json/JsonFileFeatureIO.java
+++ b/ingestion/src/main/java/feast/storage/file/json/JsonFileFeatureIO.java
@@ -20,21 +20,14 @@ package feast.storage.file.json;
 import com.google.protobuf.InvalidProtocolBufferException;
 import com.google.protobuf.Message;
 import com.google.protobuf.util.JsonFormat;
-import lombok.AllArgsConstructor;
-import lombok.Builder;
-import lombok.NonNull;
-import org.apache.beam.sdk.io.TextIO;
-import org.apache.beam.sdk.transforms.DoFn;
-import org.apache.beam.sdk.transforms.ParDo;
-import org.apache.beam.sdk.transforms.SerializableFunction;
-import org.apache.beam.sdk.values.PCollection;
-import org.apache.beam.sdk.values.PDone;
-import org.apache.beam.sdk.values.PInput;
 import feast.ingestion.transform.FeatureIO;
 import feast.storage.file.FileStoreOptions;
 import feast.storage.file.TextFileFeatureIO;
 import feast.types.FeatureRowExtendedProto.FeatureRowExtended;
-import feast.types.FeatureRowProto.FeatureRow;
+import lombok.AllArgsConstructor;
+import org.apache.beam.sdk.transforms.SerializableFunction;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.PDone;
 
 public class JsonFileFeatureIO {
 
@@ -46,43 +39,6 @@ public class JsonFileFeatureIO {
     return new Write(options, (rowExtended) -> rowExtended.getRow());
   }
 
-  /**
-   * Transform for processing JSON text jsonfiles and that contain JSON serialised FeatureRow
-   * messages, one per line.
-   *
-   * <p>This transform asserts that the import spec is for only one entity, as all columns must have
-   * the same entity.
-   *
-   * <p>The output is a PCollection of {@link feast.types.FeatureRowProto.FeatureRow
-   * FeatureRows}.
-   */
-  @Builder
-  public static class Read extends FeatureIO.Read {
-
-    @NonNull private final JsonFileSourceOptions options;
-
-    @Override
-    public PCollection<FeatureRow> expand(PInput input) {
-      PCollection<String> jsonLines = input.getPipeline().apply(TextIO.read().from(options.path));
-
-      return jsonLines.apply(
-          ParDo.of(
-              new DoFn<String, FeatureRow>() {
-                @ProcessElement
-                public void processElement(ProcessContext context) {
-                  String line = context.element();
-                  FeatureRow.Builder builder = FeatureRow.newBuilder();
-                  try {
-                    JsonFormat.parser().merge(line, builder);
-                    context.output(builder.build());
-                  } catch (InvalidProtocolBufferException e) {
-                    throw new RuntimeException(e);
-                  }
-                }
-              }));
-    }
-  }
-
   @AllArgsConstructor
   public static class Write extends FeatureIO.Write {
 
diff --git a/ingestion/src/main/java/feast/storage/file/json/JsonFileSourceOptions.java b/ingestion/src/main/java/feast/storage/file/json/JsonFileSourceOptions.java
deleted file mode 100644
index 5f805e22b..000000000
--- a/ingestion/src/main/java/feast/storage/file/json/JsonFileSourceOptions.java
+++ /dev/null
@@ -1,25 +0,0 @@
-/*
- * Copyright 2018 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.storage.file.json;
-
-import javax.validation.constraints.NotEmpty;
-import feast.options.Options;
-
-class JsonFileSourceOptions implements Options {
-  @NotEmpty String path;
-}
diff --git a/ingestion/src/main/java/feast/storage/file/json/JsonFileStores.java b/ingestion/src/main/java/feast/storage/file/json/JsonFileStores.java
index c6e70d970..11dda1e3b 100644
--- a/ingestion/src/main/java/feast/storage/file/json/JsonFileStores.java
+++ b/ingestion/src/main/java/feast/storage/file/json/JsonFileStores.java
@@ -17,40 +17,20 @@
 
 package feast.storage.file.json;
 
-import static com.google.common.base.Preconditions.checkArgument;
-
 import com.google.auto.service.AutoService;
-import lombok.AllArgsConstructor;
 import feast.ingestion.model.Specs;
-import feast.ingestion.transform.FeatureIO.Read;
 import feast.ingestion.transform.FeatureIO.Write;
 import feast.options.OptionsParser;
-import feast.specs.ImportSpecProto.ImportSpec;
 import feast.specs.StorageSpecProto.StorageSpec;
 import feast.storage.ErrorsStore;
-import feast.storage.FeatureSource;
 import feast.storage.ServingStore;
 import feast.storage.WarehouseStore;
 import feast.storage.file.FileStoreOptions;
+import lombok.AllArgsConstructor;
 
 public class JsonFileStores {
-  private static final String JSON_FILES_TYPE = "file.json";
-
-  public static class JsonFileFeatureSource implements FeatureSource {
-
-    @Override
-    public Read create(ImportSpec importSpec) {
-      checkArgument(importSpec.getType().equals(getType()));
-      JsonFileSourceOptions options =
-          OptionsParser.parse(importSpec.getOptionsMap(), JsonFileSourceOptions.class);
-      return new JsonFileFeatureIO.Read(options);
-    }
 
-    @Override
-    public String getType() {
-      return JSON_FILES_TYPE;
-    }
-  }
+  private static final String JSON_FILES_TYPE = "file.json";
 
   @AutoService(WarehouseStore.class)
   @AllArgsConstructor


Test patch:
diff --git a/core/src/test/java/feast/core/validators/SpecValidatorTest.java b/core/src/test/java/feast/core/validators/SpecValidatorTest.java
index d82f87849..0c666a11a 100644
--- a/core/src/test/java/feast/core/validators/SpecValidatorTest.java
+++ b/core/src/test/java/feast/core/validators/SpecValidatorTest.java
@@ -44,13 +44,14 @@ import org.junit.rules.ExpectedException;
 import org.mockito.Mockito;
 
 public class SpecValidatorTest {
+
+  @Rule
+  public final ExpectedException exception = ExpectedException.none();
   private FeatureInfoRepository featureInfoRepository;
   private FeatureGroupInfoRepository featureGroupInfoRepository;
   private EntityInfoRepository entityInfoRepository;
   private StorageInfoRepository storageInfoRepository;
 
-  @Rule public final ExpectedException exception = ExpectedException.none();
-
   @Before
   public void setUp() {
     featureInfoRepository = Mockito.mock(FeatureInfoRepository.class);
@@ -369,7 +370,7 @@ public class SpecValidatorTest {
     StorageInfo redis1 = new StorageInfo();
     redis1.setId(servingStoreId);
     redis1.setType("redis");
-    when(storageInfoRepository.findById( servingStoreId)).thenReturn(Optional.of(redis1));
+    when(storageInfoRepository.findById(servingStoreId)).thenReturn(Optional.of(redis1));
 
     SpecValidator validator =
         new SpecValidator(
@@ -392,7 +393,8 @@ public class SpecValidatorTest {
             .setDataStores(dataStores)
             .build();
     exception.expect(IllegalArgumentException.class);
-    exception.expectMessage(String.format("Warehouse store with id %s does not exist", warehouseStoreId));
+    exception.expectMessage(
+        String.format("Warehouse store with id %s does not exist", warehouseStoreId));
     validator.validateFeatureSpec(input);
   }
 
@@ -405,7 +407,7 @@ public class SpecValidatorTest {
     StorageInfo redis1 = new StorageInfo();
     redis1.setId(servingStoreId);
     redis1.setType("redis");
-    when(storageInfoRepository.findById( servingStoreId)).thenReturn(Optional.of(redis1));
+    when(storageInfoRepository.findById(servingStoreId)).thenReturn(Optional.of(redis1));
 
     SpecValidator validator =
         new SpecValidator(
@@ -432,18 +434,21 @@ public class SpecValidatorTest {
   @Test
   public void featureSpecWithUnsupportedWarehouseStoreShouldThrowIllegalArgumentException() {
     String servingStoreId = "REDIS1";
-    StorageSpec servingStoreSpec = StorageSpec.newBuilder().setId(servingStoreId).setType("redis").build();
+    StorageSpec servingStoreSpec = StorageSpec.newBuilder().setId(servingStoreId).setType("redis")
+        .build();
     StorageInfo servingStoreInfo = new StorageInfo(servingStoreSpec);
 
     String warehouseStoreId = "REDIS2";
-    StorageSpec warehouseStoreSpec = StorageSpec.newBuilder().setId(warehouseStoreId).setType("redis").build();
+    StorageSpec warehouseStoreSpec = StorageSpec.newBuilder().setId(warehouseStoreId)
+        .setType("redis").build();
     StorageInfo warehouseStoreInfo = new StorageInfo(warehouseStoreSpec);
 
     when(entityInfoRepository.existsById("entity")).thenReturn(true);
     when(storageInfoRepository.existsById(servingStoreId)).thenReturn(true);
     when(storageInfoRepository.existsById(warehouseStoreId)).thenReturn(true);
     when(storageInfoRepository.findById(servingStoreId)).thenReturn(Optional.of(servingStoreInfo));
-    when(storageInfoRepository.findById(warehouseStoreId)).thenReturn(Optional.of(warehouseStoreInfo));
+    when(storageInfoRepository.findById(warehouseStoreId))
+        .thenReturn(Optional.of(warehouseStoreInfo));
     SpecValidator validator =
         new SpecValidator(
             storageInfoRepository,
@@ -488,7 +493,8 @@ public class SpecValidatorTest {
     when(entityInfoRepository.existsById("entity")).thenReturn(true);
     when(storageInfoRepository.existsById(servingStoreName)).thenReturn(true);
     when(storageInfoRepository.existsById(warehouseStorageName)).thenReturn(true);
-    when(storageInfoRepository.findById(servingStoreName)).thenReturn(Optional.of(redis1StorageInfo));
+    when(storageInfoRepository.findById(servingStoreName))
+        .thenReturn(Optional.of(redis1StorageInfo));
     when(storageInfoRepository.findById(warehouseStorageName)).thenReturn(Optional.of(bqInfo));
     SpecValidator validator =
         new SpecValidator(
@@ -670,10 +676,10 @@ public class SpecValidatorTest {
             featureGroupInfoRepository,
             featureInfoRepository);
     ImportSpec input =
-        ImportSpec.newBuilder().setType("file").putOptions("format", "notSupported").build();
+        ImportSpec.newBuilder().setType("file.wat?").build();
     exception.expect(IllegalArgumentException.class);
     exception.expectMessage(
-        "Validation for import spec failed: Invalid options: File format must be of type 'json' or 'csv'");
+        "Validation for import spec failed: Type file.wat? not supported");
     validator.validateImportSpec(input);
   }
 
@@ -685,7 +691,7 @@ public class SpecValidatorTest {
             entityInfoRepository,
             featureGroupInfoRepository,
             featureInfoRepository);
-    ImportSpec input = ImportSpec.newBuilder().setType("file").putOptions("format", "csv").build();
+    ImportSpec input = ImportSpec.newBuilder().setType("file.csv").build();
     exception.expect(IllegalArgumentException.class);
     exception.expectMessage(
         "Validation for import spec failed: Invalid options: File path cannot be empty");
@@ -702,8 +708,7 @@ public class SpecValidatorTest {
             featureInfoRepository);
     ImportSpec input =
         ImportSpec.newBuilder()
-            .setType("file")
-            .putOptions("format", "csv")
+            .setType("file.csv")
             .putOptions("path", "gs://asdasd")
             .build();
     exception.expect(IllegalArgumentException.class);
@@ -779,4 +784,55 @@ public class SpecValidatorTest {
         "Validation for import spec failed: Feature some_nonexistent_feature not registered");
     validator.validateImportSpec(input);
   }
+
+  @Test
+  public void importSpecWithKafkaSourceAndCorrectOptionsShouldPassValidation() {
+    SpecValidator validator =
+        new SpecValidator(
+            storageInfoRepository,
+            entityInfoRepository,
+            featureGroupInfoRepository,
+            featureInfoRepository);
+    when(featureInfoRepository.existsById("some_existing_feature")).thenReturn(true);
+    when(entityInfoRepository.existsById("someEntity")).thenReturn(true);
+    Schema schema =
+        Schema.newBuilder()
+            .addFields(Field.newBuilder().setFeatureId("some_existing_feature").build())
+            .build();
+    ImportSpec input =
+        ImportSpec.newBuilder()
+            .setType("kafka")
+            .putOptions("topics", "my-kafka-topic")
+            .putOptions("server", "localhost:54321")
+            .setSchema(schema)
+            .addEntities("someEntity")
+            .build();
+    validator.validateImportSpec(input);
+  }
+
+  @Test
+  public void importSpecWithKafkaSourceWithoutOptionsShouldThrowIllegalArgumentException() {
+    SpecValidator validator =
+        new SpecValidator(
+            storageInfoRepository,
+            entityInfoRepository,
+            featureGroupInfoRepository,
+            featureInfoRepository);
+    when(featureInfoRepository.existsById("some_existing_feature")).thenReturn(true);
+    when(entityInfoRepository.existsById("someEntity")).thenReturn(true);
+    Schema schema =
+        Schema.newBuilder()
+            .addFields(Field.newBuilder().setFeatureId("some_existing_feature").build())
+            .build();
+    ImportSpec input =
+        ImportSpec.newBuilder()
+            .setType("kafka")
+            .setSchema(schema)
+            .addEntities("someEntity")
+            .build();
+    exception.expect(IllegalArgumentException.class);
+    exception.expectMessage(
+        "Validation for import spec failed: Invalid options: Kafka ingestion requires either topics or servers");
+    validator.validateImportSpec(input);
+  }
 }
diff --git a/ingestion/src/test/java/feast/ingestion/ImportJobCSVTest.java b/ingestion/src/test/java/feast/ingestion/ImportJobCSVTest.java
index 61f44f4d4..c6adf7dd2 100644
--- a/ingestion/src/test/java/feast/ingestion/ImportJobCSVTest.java
+++ b/ingestion/src/test/java/feast/ingestion/ImportJobCSVTest.java
@@ -89,9 +89,8 @@ public class ImportJobCSVTest {
     ImportSpec importSpec =
         ProtoUtil.decodeProtoYaml(
             "---\n"
-                + "type: file\n"
+                + "type: file.csv\n"
                 + "options:\n"
-                + "  format: csv\n"
                 + "  # path: to be overwritten in tests\n"
                 + "entities:\n"
                 + "  - testEntity\n"
@@ -179,7 +178,7 @@ public class ImportJobCSVTest {
     ImportSpec importSpec =
         ProtoUtil.decodeProtoYaml(
             "---\n"
-                + "type: file\n"
+                + "type: file.csv\n"
                 + "options:\n"
                 + "  format: csv\n"
                 + "  # path: to be overwritten in tests\n"
@@ -190,7 +189,8 @@ public class ImportJobCSVTest {
                 + "  timestampValue: 2018-09-25T00:00:00.000Z\n"
                 + "  fields:\n"
                 + "    - name: id\n"
-                + "    - featureId: testEntity.none.unknownInt32\n" // Unknown store is not available
+                + "    - featureId: testEntity.none.unknownInt32\n"
+                // Unknown store is not available
                 + "    - featureId: testEntity.none.testString\n"
                 + "\n",
             ImportSpec.getDefaultInstance());
@@ -217,9 +217,8 @@ public class ImportJobCSVTest {
     ImportSpec importSpec =
         ProtoUtil.decodeProtoYaml(
             "---\n"
-                + "type: file\n"
+                + "type: file.csv\n"
                 + "options:\n"
-                + "  format: csv\n"
                 + "  # path: to be overwritten in tests\n"
                 + "entities:\n"
                 + "  - testEntity\n"
@@ -283,9 +282,8 @@ public class ImportJobCSVTest {
     ImportSpec importSpec =
         ProtoUtil.decodeProtoYaml(
             "---\n"
-                + "type: file\n"
+                + "type: file.csv\n"
                 + "options:\n"
-                + "  format: csv\n"
                 + "  # path: to be overwritten in tests\n"
                 + "entities:\n"
                 + "  - testEntity\n"
diff --git a/ingestion/src/test/java/feast/ingestion/model/SpecsTest.java b/ingestion/src/test/java/feast/ingestion/model/SpecsTest.java
index 17aceba18..d07073105 100644
--- a/ingestion/src/test/java/feast/ingestion/model/SpecsTest.java
+++ b/ingestion/src/test/java/feast/ingestion/model/SpecsTest.java
@@ -1,3 +1,20 @@
+/*
+ * 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
+ *
+ * http://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.ingestion.model;
 
 import static junit.framework.TestCase.assertTrue;
diff --git a/ingestion/src/test/java/feast/ingestion/service/MockSpecService.java b/ingestion/src/test/java/feast/ingestion/service/MockSpecService.java
index 626545e11..54e70b39a 100644
--- a/ingestion/src/test/java/feast/ingestion/service/MockSpecService.java
+++ b/ingestion/src/test/java/feast/ingestion/service/MockSpecService.java
@@ -1,3 +1,20 @@
+/*
+ * 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
+ *
+ * http://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.ingestion.service;
 
 import com.google.common.collect.Maps;
diff --git a/ingestion/src/test/java/feast/ingestion/transform/SplitOutputByStoreTest.java b/ingestion/src/test/java/feast/ingestion/transform/SplitOutputByStoreTest.java
index 09e8ef543..7758a7748 100644
--- a/ingestion/src/test/java/feast/ingestion/transform/SplitOutputByStoreTest.java
+++ b/ingestion/src/test/java/feast/ingestion/transform/SplitOutputByStoreTest.java
@@ -1,3 +1,20 @@
+/*
+ * 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
+ *
+ * http://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.ingestion.transform;
 
 import static junit.framework.TestCase.assertNull;
diff --git a/ingestion/src/test/java/feast/storage/bigquery/FeatureRowFromBigQuerySchemaAndRecordFnTest.java b/ingestion/src/test/java/feast/source/bigquery/BQToFeatureRowFnTest.java
similarity index 94%
rename from ingestion/src/test/java/feast/storage/bigquery/FeatureRowFromBigQuerySchemaAndRecordFnTest.java
rename to ingestion/src/test/java/feast/source/bigquery/BQToFeatureRowFnTest.java
index 433cdecf5..4c809fb5c 100644
--- a/ingestion/src/test/java/feast/storage/bigquery/FeatureRowFromBigQuerySchemaAndRecordFnTest.java
+++ b/ingestion/src/test/java/feast/source/bigquery/BQToFeatureRowFnTest.java
@@ -15,7 +15,7 @@
  *
  */
 
-package feast.storage.bigquery;
+package feast.source.bigquery;
 
 import static org.hamcrest.Matchers.equalTo;
 import static org.mockito.Mockito.mock;
@@ -26,6 +26,7 @@ import com.google.api.services.bigquery.model.TableSchema;
 import com.google.cloud.bigquery.LegacySQLTypeName;
 import com.google.common.collect.Lists;
 import com.google.protobuf.Timestamp;
+import feast.source.bigquery.BigQueryToFeatureRowFn;
 import org.apache.avro.generic.GenericRecord;
 import org.apache.beam.sdk.io.gcp.bigquery.SchemaAndRecord;
 import org.joda.time.DateTime;
@@ -39,7 +40,7 @@ import feast.specs.ImportSpecProto.ImportSpec;
 import feast.specs.ImportSpecProto.Schema;
 import feast.types.FeatureRowProto.FeatureRow;
 
-public class FeatureRowFromBigQuerySchemaAndRecordFnTest {
+public class BQToFeatureRowFnTest {
   @Test
   public void testImportSpecFieldsMissingFromBQTable() {
     // TODO what if a field in the import spec is not in the bq schema
@@ -77,7 +78,7 @@ public class FeatureRowFromBigQuerySchemaAndRecordFnTest {
             new TableFieldSchema().setName("bq_timestamp").setType(LegacySQLTypeName.TIMESTAMP.name()),
             new TableFieldSchema().setName("bq_value").setType(LegacySQLTypeName.INTEGER.name())));
     SchemaAndRecord schemaAndRecord = new SchemaAndRecord(record, tableSchema);
-    FeatureRow row = new FeatureRowFromBigQuerySchemaAndRecordFn(importSpec).apply(schemaAndRecord);
+    FeatureRow row = new BigQueryToFeatureRowFn(importSpec).apply(schemaAndRecord);
     Assert.assertEquals(now, row.getEventTimestamp());
     Assert.assertEquals("abcd", row.getEntityKey());
     Assert.assertEquals("testEntity", row.getEntityName());
@@ -120,7 +121,7 @@ public class FeatureRowFromBigQuerySchemaAndRecordFnTest {
             new TableFieldSchema().setName("bq_timestamp").setType(LegacySQLTypeName.TIMESTAMP.name()),
             new TableFieldSchema().setName("bq_value").setType(LegacySQLTypeName.INTEGER.name())));
     SchemaAndRecord schemaAndRecord = new SchemaAndRecord(record, tableSchema);
-    FeatureRow row = new FeatureRowFromBigQuerySchemaAndRecordFn(importSpec).apply(schemaAndRecord);
+    FeatureRow row = new BigQueryToFeatureRowFn(importSpec).apply(schemaAndRecord);
     Assert.assertEquals(now, row.getEventTimestamp());
     Assert.assertEquals("1234", row.getEntityKey());
     Assert.assertEquals("testEntity", row.getEntityName());
diff --git a/ingestion/src/test/java/feast/storage/bigquery/BigQuerySourceOptionsTest.java b/ingestion/src/test/java/feast/source/bigquery/BigQuerySourceOptionsTest.java
similarity index 95%
rename from ingestion/src/test/java/feast/storage/bigquery/BigQuerySourceOptionsTest.java
rename to ingestion/src/test/java/feast/source/bigquery/BigQuerySourceOptionsTest.java
index d2e87dc3a..27b9842e0 100644
--- a/ingestion/src/test/java/feast/storage/bigquery/BigQuerySourceOptionsTest.java
+++ b/ingestion/src/test/java/feast/source/bigquery/BigQuerySourceOptionsTest.java
@@ -15,11 +15,12 @@
  *
  */
 
-package feast.storage.bigquery;
+package feast.source.bigquery;
 
+import feast.options.OptionsParser;
+import feast.source.bigquery.BigQueryFeatureSource.BigQuerySourceOptions;
 import org.junit.Assert;
 import org.junit.Test;
-import feast.options.OptionsParser;
 import org.testcontainers.shaded.com.google.common.collect.ImmutableMap;
 
 public class BigQuerySourceOptionsTest {
diff --git a/ingestion/src/test/java/feast/ingestion/transform/CsvIOTest.java b/ingestion/src/test/java/feast/source/csv/CsvIOTest.java
similarity index 87%
rename from ingestion/src/test/java/feast/ingestion/transform/CsvIOTest.java
rename to ingestion/src/test/java/feast/source/csv/CsvIOTest.java
index aefe5f00e..5e5217af9 100644
--- a/ingestion/src/test/java/feast/ingestion/transform/CsvIOTest.java
+++ b/ingestion/src/test/java/feast/source/csv/CsvIOTest.java
@@ -15,18 +15,16 @@
  *
  */
 
-package feast.ingestion.transform;
+package feast.source.csv;
 
 import static org.junit.Assert.*;
 import static org.hamcrest.CoreMatchers.*;
 
 import com.google.common.collect.Lists;
-import feast.ingestion.transform.CsvIO.CSVLineParser;
-import feast.ingestion.transform.CsvIO.StringMap;
+import feast.source.csv.CsvIO.CSVLineParser;
+import feast.source.csv.CsvIO.StringMap;
 import java.util.List;
 import junit.framework.TestCase;
-import feast.ingestion.transform.CsvIO.CSVLineParser;
-import feast.ingestion.transform.CsvIO.StringMap;
 
 public class CsvIOTest extends TestCase {
 
diff --git a/ingestion/src/test/java/feast/ingestion/deserializer/KafkaFeatureRowDeserializerTest.java b/ingestion/src/test/java/feast/source/kafka/deserializer/KafkaFeatureRowDeserializerTest.java
similarity index 99%
rename from ingestion/src/test/java/feast/ingestion/deserializer/KafkaFeatureRowDeserializerTest.java
rename to ingestion/src/test/java/feast/source/kafka/deserializer/KafkaFeatureRowDeserializerTest.java
index 4b9ece362..3436b8d8b 100644
--- a/ingestion/src/test/java/feast/ingestion/deserializer/KafkaFeatureRowDeserializerTest.java
+++ b/ingestion/src/test/java/feast/source/kafka/deserializer/KafkaFeatureRowDeserializerTest.java
@@ -15,7 +15,7 @@
  *
  */
 
-package feast.ingestion.deserializer;
+package feast.source.kafka.deserializer;
 
 import com.google.protobuf.MessageLite;
 import feast.types.FeatureRowProto.FeatureRow;