Report observed value for aggregated checks in pre-ingestion feature validation (feast-dev#1278)

pyalex · web-flow · commit dd36544b03db · 2021-01-20T12:08:23.000+08:00
* report observed value for aggregation checks

Signed-off-by: Oleksii Moskalenko &lt;moskalenko.alexey@gmail.com&gt;

* fix kubectl version in ci docker

Signed-off-by: Oleksii Moskalenko &lt;moskalenko.alexey@gmail.com&gt;

* fix kubectl version in ci docker

Signed-off-by: Oleksii Moskalenko &lt;moskalenko.alexey@gmail.com&gt;

* fix kubectl version in ci docker

Signed-off-by: Oleksii Moskalenko &lt;moskalenko.alexey@gmail.com&gt;
diff --git a/infra/docker/ci/Dockerfile b/infra/docker/ci/Dockerfile
@@ -77,7 +77,7 @@ RUN PROTOC_ZIP=protoc-${PROTOC_VERSION}-linux-x86_64.zip && \
 RUN curl -sL https://aka.ms/InstallAzureCLIDeb | bash
 
 # Install kubectl
-RUN apt-get install -y kubectl=1.20.1-00
+RUN apt-get install -y kubectl=1.20.2-00
 
 # Install helm
 RUN curl -fsSL -o get_helm.sh https://raw.githubusercontent.com/helm/helm/master/scripts/get-helm-3 && \
diff --git a/sdk/python/feast/contrib/validation/ge.py b/sdk/python/feast/contrib/validation/ge.py
@@ -109,11 +109,9 @@ def udf(df: pd.DataFrame) -> pd.Series:
             if check.success:
                 continue
 
-            unexpected_count = (
-                check.result["unexpected_count"]
-                if "unexpected_count" in check.result
-                else df.shape[0]
-            )
+            if check.exception_info["raised_exception"]:
+                # ToDo: probably we should mark all rows as invalid
+                continue
 
             check_kwargs = check.expectation_config.kwargs
             check_kwargs.pop("result_format", None)
@@ -126,21 +124,31 @@ def udf(df: pd.DataFrame) -> pd.Series:
                 ]
             )
 
-            reporter.increment(
-                "feast_feature_validation_check_failed",
-                value=unexpected_count,
-                tags=[
-                    f"feature_table:{os.getenv('FEAST_INGESTION_FEATURE_TABLE', 'unknown')}",
-                    f"project:{os.getenv('FEAST_INGESTION_PROJECT_NAME', 'default')}",
-                    f"check:{check_name}",
-                ],
-            )
-
-            if check.exception_info["raised_exception"]:
-                # ToDo: probably we should mark all rows as invalid
-                continue
+            if "unexpected_count" in check.result:
+                reporter.increment(
+                    "feast_feature_validation_check_failed",
+                    value=check.result["unexpected_count"],
+                    tags=[
+                        f"feature_table:{os.getenv('FEAST_INGESTION_FEATURE_TABLE', 'unknown')}",
+                        f"project:{os.getenv('FEAST_INGESTION_PROJECT_NAME', 'default')}",
+                        f"check:{check_name}",
+                    ],
+                )
 
-            valid_rows.iloc[check.result["unexpected_index_list"]] = False
+                valid_rows.iloc[check.result["unexpected_index_list"]] = False
+
+            elif "observed_value" in check.result:
+                reporter.increment(
+                    "feast_feature_validation_observed_value",
+                    value=int(
+                        check.result["observed_value"] * 100
+                    ),  # storing as decimal with precision 2
+                    tags=[
+                        f"feature_table:{os.getenv('FEAST_INGESTION_FEATURE_TABLE', 'unknown')}",
+                        f"project:{os.getenv('FEAST_INGESTION_PROJECT_NAME', 'default')}",
+                        f"check:{check_name}",
+                    ],
+                )
 
         return valid_rows
 
diff --git a/sdk/python/feast/job_service.py b/sdk/python/feast/job_service.py
@@ -8,6 +8,7 @@
 from typing import Dict, List, Tuple, cast
 
 import grpc
+from google.api_core.exceptions import FailedPrecondition
 from google.protobuf.timestamp_pb2 import Timestamp
 
 import feast
@@ -334,7 +335,10 @@ def ensure_stream_ingestion_jobs(client: feast.Client, all_projects: bool):
         logging.info(
             f"Cancelling a stream ingestion job with job_hash={job_hash} job_id={job.get_id()} status={job.get_status()}"
         )
-        job.cancel()
+        try:
+            job.cancel()
+        except FailedPrecondition as exc:
+            logging.warning(f"Job canceling failed with exception {exc}")
 
     for job_hash in job_hashes_to_start:
         # Any job that we wish to start should be among expected table refs map
diff --git a/sdk/python/feast/staging/entities.py b/sdk/python/feast/staging/entities.py
@@ -123,7 +123,7 @@ def create_bq_view_of_joined_features_and_entities(
     view.view_query = JOIN_TEMPLATE.format(
         entities=entities_ref,
         source=source_ref,
-        entity_key=",".join([f"source.{e} = entities.{e}" for e in entity_names]),
+        entity_key=" AND ".join([f"source.{e} = entities.{e}" for e in entity_names]),
     )
     view.expires = datetime.now() + timedelta(days=1)
     bq_client.create_table(view)

Original file line number	Diff line number	Diff line change
`@@ -123,7 +123,7 @@ def create_bq_view_of_joined_features_and_entities(`
`123`	`123`	`view.view_query = JOIN_TEMPLATE.format(`
`124`	`124`	`entities=entities_ref,`
`125`	`125`	`source=source_ref,`
`126`		`- entity_key=",".join([f"source.{e} = entities.{e}" for e in entity_names]),`
	`126`	`+ entity_key=" AND ".join([f"source.{e} = entities.{e}" for e in entity_names]),`
`127`	`127`	`)`
`128`	`128`	`view.expires = datetime.now() + timedelta(days=1)`
`129`	`129`	`bq_client.create_table(view)`