feast-dev · feast-ci-bot · Jan 17, 2019 · Dec 18, 2018 · Dec 18, 2018 · Dec 19, 2018
diff --git a/charts/feast/templates/core-deploy.yaml b/charts/feast/templates/core-deploy.yaml
@@ -69,6 +69,10 @@ spec:
               secretKeyRef:
                 name: {{ template "postgresql.fullname" . }}
                 key: postgres-password
+          - name: PROJECT_ID
+            value: "{{ .Values.core.projectId}}"
+          - name: TRAINING_DATASET_PREFIX
+            value: "{{ .Values.core.trainingDatasetPrefix }}"
           - name: CORE_API_URI
             value: {{ .Values.core.service.extIPAdr }}:{{ .Values.core.service.grpc.targetPort }}
           - name: JOB_RUNNER

diff --git a/charts/feast/values.yaml b/charts/feast/values.yaml
@@ -1,5 +1,6 @@
 --- 
-core: 
+core:
+  projectId: "gcp-project-id"
   image: 
     pullPolicy: IfNotPresent
     registry: feast
@@ -29,7 +30,8 @@ core:
     errorsStoreId: "STDOUT"
     monitoring:
       period: 5000
-      initialDelay: 60000      
+      initialDelay: 60000
+  trainingDatasetPrefix: "fs"
 
 postgresql: 
   name: feast-metadata

diff --git a/core/pom.xml b/core/pom.xml
@@ -79,12 +79,6 @@
             <artifactId>spring-boot-starter-data-jpa</artifactId>
             <version>${springBootVersion}</version>
         </dependency>
-        <!--compile "org.springframework.data:spring-data-jpa:${springBootVersion}"-->
-        <dependency>
-            <groupId>org.springframework.data</groupId>
-            <artifactId>spring-data-jpa</artifactId>
-            <version>${springBootVersion}</version>
-        </dependency>
         <!--compile "org.springframework.boot:spring-boot-starter-actuator:${springBootVersion}"-->
         <dependency>
             <groupId>org.springframework.boot</groupId>
@@ -196,6 +190,11 @@
             <artifactId>compiler</artifactId>
             <version>0.9.5</version>
         </dependency>
+        <dependency>
+            <groupId>com.hubspot.jinjava</groupId>
+            <artifactId>jinjava</artifactId>
+            <version>2.4.12</version>
+        </dependency>
         <!--compile 'io.micrometer:micrometer-core:1.0.7'-->
         <dependency>
             <groupId>io.micrometer</groupId>

diff --git a/core/src/main/java/feast/core/config/TrainingConfig.java b/core/src/main/java/feast/core/config/TrainingConfig.java
@@ -0,0 +1,43 @@
+package feast.core.config;
+
+import com.google.cloud.bigquery.BigQuery;
+import com.google.cloud.bigquery.BigQueryOptions;
+import com.google.common.base.Charsets;
+import com.google.common.io.CharStreams;
+import com.hubspot.jinjava.Jinjava;
+import feast.core.dao.FeatureInfoRepository;
+import feast.core.training.BigQueryDatasetCreator;
+import feast.core.training.BigQueryDatasetTemplater;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.time.Clock;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.core.io.ClassPathResource;
+import org.springframework.core.io.Resource;
+
+/** Configuration related to training API */
+@Configuration
+public class TrainingConfig {
+
+  @Bean
+  public BigQueryDatasetTemplater getBigQueryTrainingDatasetTemplater(
+      FeatureInfoRepository featureInfoRepository) throws IOException {
+    Resource resource = new ClassPathResource("templates/bq_training.tmpl");
+    InputStream resourceInputStream = resource.getInputStream();
+    String tmpl = CharStreams.toString(new InputStreamReader(resourceInputStream, Charsets.UTF_8));
+    return new BigQueryDatasetTemplater(new Jinjava(), tmpl, featureInfoRepository);
+  }
+
+  @Bean
+  public BigQueryDatasetCreator getBigQueryTrainingDatasetCreator(
+      BigQueryDatasetTemplater templater,
+      @Value("${feast.core.projectId}") String projectId,
+      @Value("${feast.core.datasetPrefix}") String datasetPrefix) {
+    BigQuery bigquery = BigQueryOptions.newBuilder().setProjectId(projectId).build().getService();
+    Clock clock = Clock.systemUTC();
+    return new BigQueryDatasetCreator(templater, bigquery, clock, projectId, datasetPrefix);
+  }
+}
diff --git a/core/src/main/java/feast/core/exception/TrainingDatasetCreationException.java b/core/src/main/java/feast/core/exception/TrainingDatasetCreationException.java
@@ -0,0 +1,18 @@
+package feast.core.exception;
+
+/**
+ * Exception that happens when creation of training dataset failed.
+ */
+public class TrainingDatasetCreationException extends RuntimeException {
+  public TrainingDatasetCreationException() {
+    super();
+  }
+
+  public TrainingDatasetCreationException(String message) {
+    super(message);
+  }
+
+  public TrainingDatasetCreationException(String message, Throwable cause) {
+    super(message, cause);
+  }
+}
diff --git a/core/src/main/java/feast/core/grpc/DatasetServiceImpl.java b/core/src/main/java/feast/core/grpc/DatasetServiceImpl.java
@@ -0,0 +1,120 @@
+/*
+ * 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.core.grpc;
+
+import com.google.common.base.Strings;
+import com.google.protobuf.Timestamp;
+import feast.core.DatasetServiceGrpc.DatasetServiceImplBase;
+import feast.core.DatasetServiceProto.DatasetInfo;
+import feast.core.DatasetServiceProto.FeatureSet;
+import feast.core.DatasetServiceProto.DatasetServiceTypes.CreateDatasetRequest;
+import feast.core.DatasetServiceProto.DatasetServiceTypes.CreateDatasetResponse;
+import feast.core.training.BigQueryDatasetCreator;
+import io.grpc.Status;
+import io.grpc.Status.Code;
+import io.grpc.stub.StreamObserver;
+import java.time.Instant;
+import java.time.temporal.ChronoUnit;
+import lombok.extern.slf4j.Slf4j;
+import org.lognet.springboot.grpc.GRpcService;
+import org.springframework.beans.factory.annotation.Autowired;
+
+@Slf4j
+@GRpcService
+public class DatasetServiceImpl extends DatasetServiceImplBase {
+
+  private final BigQueryDatasetCreator datasetCreator;
+
+  @Autowired
+  public DatasetServiceImpl(BigQueryDatasetCreator DatasetCreator) {
+    this.datasetCreator = DatasetCreator;
+  }
+
+  @Override
+  public void createDataset(
+      CreateDatasetRequest request,
+      StreamObserver<CreateDatasetResponse> responseObserver) {
+    try {
+      checkRequest(request);
+    } catch (IllegalArgumentException e) {
+      responseObserver.onError(
+          Status.fromCode(Code.INVALID_ARGUMENT)
+              .withCause(e)
+              .withDescription(e.getMessage())
+              .asException());
+      return;
+    }
+
+    try {
+      DatasetInfo datasetInfo =
+          datasetCreator.createDataset(
+              request.getFeatureSet(),
+              request.getStartDate(),
+              request.getEndDate(),
+              request.getLimit(),
+              request.getNamePrefix());
+      CreateDatasetResponse response =
+          CreateDatasetResponse.newBuilder().setDatasetInfo(datasetInfo).build();
+
+      responseObserver.onNext(response);
+      responseObserver.onCompleted();
+    } catch (Exception e) {
+      log.error("Training dataset creation failed", e);
+      responseObserver.onError(
+          Status.fromCode(Code.INTERNAL)
+              .withCause(e)
+              .withDescription("Training dataset creation failed: " + e.getMessage())
+              .asException());
+    }
+  }
+
+  private void checkRequest(CreateDatasetRequest request) {
+    FeatureSet featureSet = request.getFeatureSet();
+    Timestamp startDate = request.getStartDate();
+    Timestamp endDate = request.getEndDate();
+
+    checkHasSameEntity(featureSet);
+    checkStartIsBeforeEnd(startDate, endDate);
+  }
+
+  private void checkStartIsBeforeEnd(Timestamp startDate, Timestamp endDate) {
+    Instant start = Instant.ofEpochSecond(startDate.getSeconds()).truncatedTo(ChronoUnit.DAYS);
+    Instant end = Instant.ofEpochSecond(endDate.getSeconds()).truncatedTo(ChronoUnit.DAYS);
+
+    if (start.compareTo(end) > 0) {
+      throw new IllegalArgumentException("startDate is after endDate");
+    }
+  }
+
+  private void checkHasSameEntity(FeatureSet featureSet) {
+    String entityName = featureSet.getEntityName();
+    if (Strings.isNullOrEmpty(entityName)) {
+      throw new IllegalArgumentException("entity name in feature set is null or empty");
+    }
+
+    if (featureSet.getFeatureIdsCount() < 1) {
+      throw new IllegalArgumentException("feature set is empty");
+    }
+
+    for (String featureId : featureSet.getFeatureIdsList()) {
+      String entity = featureId.split("\\.")[0];
+      if (!entityName.equals(entity)) {
+        throw new IllegalArgumentException("feature set contains different entity name: " + entity);
+      }
+    }
+  }
+}
diff --git a/core/src/main/java/feast/core/training/BigQueryDatasetCreator.java b/core/src/main/java/feast/core/training/BigQueryDatasetCreator.java
@@ -0,0 +1,147 @@
+/*
+ * 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.core.training;
+
+import com.google.cloud.bigquery.BigQuery;
+import com.google.cloud.bigquery.BigQuery.JobOption;
+import com.google.cloud.bigquery.JobException;
+import com.google.cloud.bigquery.QueryJobConfiguration;
+import com.google.cloud.bigquery.TableId;
+import com.google.common.base.Strings;
+import com.google.protobuf.Timestamp;
+import feast.core.DatasetServiceProto.DatasetInfo;
+import feast.core.DatasetServiceProto.FeatureSet;
+import feast.core.exception.TrainingDatasetCreationException;
+import java.time.Clock;
+import java.time.Instant;
+import java.time.ZoneId;
+import java.time.format.DateTimeFormatter;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class BigQueryDatasetCreator {
+
+  private final String projectId;
+  private final String datasetPrefix;
+  private final BigQueryDatasetTemplater templater;
+  private final BigQuery bigQuery;
+  private final DateTimeFormatter formatter;
+  private final Clock clock;
+
+  public BigQueryDatasetCreator(
+      BigQueryDatasetTemplater templater,
+      BigQuery bigQuery,
+      Clock clock,
+      String projectId,
+      String datasetPrefix) {
+    this.templater = templater;
+    this.projectId = projectId;
+    this.datasetPrefix = datasetPrefix;
+    this.bigQuery = bigQuery;
+    this.clock = clock;
+    this.formatter = DateTimeFormatter.ofPattern("yyyyMMdd").withZone(ZoneId.of("UTC"));
+  }
+
+  /**
+   * Create dataset for a feature set
+   *
+   * @param featureSet feature set for which the training dataset should be created
+   * @param startDate starting date of the training dataset (inclusive)
+   * @param endDate end date of the training dataset (inclusive)
+   * @param limit maximum number of row should be created.
+   * @param namePrefix prefix for dataset name
+   * @return dataset info associated with the created training dataset
+   */
+  public DatasetInfo createDataset(
+      FeatureSet featureSet,
+      Timestamp startDate,
+      Timestamp endDate,
+      long limit,
+      String namePrefix) {
+    try {
+      String query = templater.createQuery(featureSet, startDate, endDate, limit);
+      String tableName = createBqTableName(startDate, endDate, namePrefix);
+      String bqDatasetName = createBqDatasetName(featureSet.getEntityName());
+
+      createBqDatasetIfMissing(bqDatasetName);
+
+      TableId destinationTable =
+          TableId.of(projectId, createBqDatasetName(featureSet.getEntityName()), tableName);
+      QueryJobConfiguration queryConfig =
+          QueryJobConfiguration.newBuilder(query)
+              .setAllowLargeResults(true)
+              .setDestinationTable(destinationTable)
+              .build();
+      JobOption jobOption = JobOption.fields();
+      bigQuery.query(queryConfig, jobOption);
+      return DatasetInfo.newBuilder()
+          .setName(createTrainingDatasetName(namePrefix, featureSet.getEntityName(), tableName))
+          .setTableUrl(toTableUrl(destinationTable))
+          .build();
+    } catch (JobException e) {
+      log.error("Failed creating training dataset", e);
+      throw new TrainingDatasetCreationException("Failed creating training dataset", e);
+    } catch (InterruptedException e) {
+      log.error("Training dataset creation was interrupted", e);
+      throw new TrainingDatasetCreationException("Training dataset creation was interrupted", e);
+    }
+  }
+
+  private void createBqDatasetIfMissing(String bqDatasetName) {
+    if (bigQuery.getDataset(bqDatasetName) != null) {
+      return;
+    }
+
+    // create dataset
+    bigQuery.create(com.google.cloud.bigquery.DatasetInfo.of(bqDatasetName));
+  }
+
+  private String createBqTableName(Timestamp startDate, Timestamp endDate, String namePrefix) {
+    String currentTime = String.valueOf(clock.millis());
+    if (!Strings.isNullOrEmpty(namePrefix)) {
+      //  only alphanumeric and underscore are allowed
+      namePrefix = namePrefix.replaceAll("[^a-zA-Z0-9_]", "_");
+      return String.format(
+          "%s_%s_%s_%s",
+          namePrefix, currentTime, formatTimestamp(startDate), formatTimestamp(endDate));
+    }
+
+    return String.format(
+        "%s_%s_%s", currentTime, formatTimestamp(startDate), formatTimestamp(endDate));
+  }
+
+  private String createBqDatasetName(String entity) {
+    return String.format("%s_%s", datasetPrefix, entity);
+  }
+
+  private String formatTimestamp(Timestamp timestamp) {
+    Instant instant = Instant.ofEpochSecond(timestamp.getSeconds());
+    return formatter.format(instant);
+  }
+
+  private String toTableUrl(TableId tableId) {
+    return String.format(
+        "%s.%s.%s", tableId.getProject(), tableId.getDataset(), tableId.getTable());
+  }
+
+  private String createTrainingDatasetName(String namePrefix, String entityName, String tableName) {
+    if (!Strings.isNullOrEmpty(namePrefix)) {
+      return tableName;
+    }
+    return String.format("%s_%s", entityName, tableName);
+  }
+}