feat(ingestion): extend feast plugin to ingest tags and owners

metadata-ingestion/scripts/datahub_preflight.sh

@@ -1,4 +1,4 @@
-#!/bin/bash -e
+ #!/bin/bash -e
 
 #From https://stackoverflow.com/questions/4023830/how-to-compare-two-strings-in-dot-separated-version-format-in-bash
 verlte() {
@@ -45,7 +45,7 @@ arm64_darwin_preflight() {
   	pip3 install --no-use-pep517 scipy
   fi
 
-  brew_install "openssl@1.1"
+  brew_install "openssl@3.0.14"
   brew install "postgresql@14"
 
   # postgresql installs libs in a strange way

metadata-ingestion/src/datahub/ingestion/source/feast.py

@@ -33,6 +33,11 @@
 )
 from datahub.ingestion.api.source import Source, SourceReport
 from datahub.ingestion.api.workunit import MetadataWorkUnit
+from datahub.metadata._schema_classes import (
+    OwnerClass,
+    OwnershipClass,
+    OwnershipTypeClass,
+)
 from datahub.metadata.com.linkedin.pegasus2avro.common import MLFeatureDataType
 from datahub.metadata.com.linkedin.pegasus2avro.metadata.snapshot import (
     MLFeatureSnapshot,
@@ -42,10 +47,12 @@
 from datahub.metadata.com.linkedin.pegasus2avro.mxe import MetadataChangeEvent
 from datahub.metadata.schema_classes import (
     BrowsePathsClass,
+    GlobalTagsClass,
     MLFeaturePropertiesClass,
     MLFeatureTablePropertiesClass,
     MLPrimaryKeyPropertiesClass,
     StatusClass,
+    TagAssociationClass,
 )
 
 # FIXME: ValueType module cannot be used as a type
@@ -216,9 +223,26 @@ def _get_entity_workunit(
 
         feature_view_name = f"{self.feature_store.project}.{feature_view.name}"
 
+        aspects = [StatusClass(removed=False)]
+
+        if entity.tags.get("name"):
+            tag: str = entity.tags.get("name")
+            tag_association = TagAssociationClass(tag=builder.make_tag_urn(tag))
+            global_tags_aspect = GlobalTagsClass(tags=[tag_association])
+            aspects.append(global_tags_aspect)
+
+        if entity.owner:
+            owner = entity.owner
+            owner_association = OwnerClass(
+                owner=builder.make_owner_urn(owner, owner_type=builder.OwnerType.USER),
+                type=OwnershipTypeClass.TECHNICAL_OWNER,
+            )
+            owners_aspect = OwnershipClass(owners=[owner_association])
+            aspects.append(owners_aspect)
+
         entity_snapshot = MLPrimaryKeySnapshot(
             urn=builder.make_ml_primary_key_urn(feature_view_name, entity.name),
-            aspects=[StatusClass(removed=False)],
+            aspects=aspects,
         )
 
         entity_snapshot.aspects.append(
@@ -243,10 +267,20 @@ def _get_feature_workunit(
         Generate an MLFeature work unit for a Feast feature.
         """
         feature_view_name = f"{self.feature_store.project}.{feature_view.name}"
+        global_tags_aspect = None
+
+        if field.tags.get("name"):
+            tag_name = field.tags.get("name")
+            tag_association = TagAssociationClass(tag=builder.make_tag_urn(tag_name))
+            global_tags_aspect = GlobalTagsClass(tags=[tag_association])
+
+        aspects = [StatusClass(removed=False)]
+        if global_tags_aspect is not None:
+            aspects.append(global_tags_aspect)
 
         feature_snapshot = MLFeatureSnapshot(
             urn=builder.make_ml_feature_urn(feature_view_name, field.name),
-            aspects=[StatusClass(removed=False)],
+            aspects=aspects,
         )
 
         feature_sources = []
@@ -296,12 +330,29 @@ def _get_feature_view_workunit(self, feature_view: FeatureView) -> MetadataWorkU
 
         feature_view_name = f"{self.feature_store.project}.{feature_view.name}"
 
+        aspects = [
+            BrowsePathsClass(paths=[f"/feast/{self.feature_store.project}"]),
+            StatusClass(removed=False),
+        ]
+
+        if feature_view.tags.get("name"):
+            tag = feature_view.tags.get("name")
+            tag_association = TagAssociationClass(tag=builder.make_tag_urn(tag))
+            global_tags_aspect = GlobalTagsClass(tags=[tag_association])
+            aspects.append(global_tags_aspect)
+
+        if feature_view.owner:
+            owner = feature_view.owner
+            owner_association = OwnerClass(
+                owner=builder.make_owner_urn(owner, owner_type=builder.OwnerType.USER),
+                type=OwnershipTypeClass.TECHNICAL_OWNER,
+            )
+            owners_aspect = OwnershipClass(owners=[owner_association])
+            aspects.append(owners_aspect)
+
         feature_view_snapshot = MLFeatureTableSnapshot(
             urn=builder.make_ml_feature_table_urn("feast", feature_view_name),
-            aspects=[
-                BrowsePathsClass(paths=[f"/feast/{self.feature_store.project}"]),
-                StatusClass(removed=False),
-            ],
+            aspects=aspects,
         )
 
         feature_view_snapshot.aspects.append(
@@ -366,6 +417,7 @@ def create(cls, config_dict, ctx):
         return cls(config, ctx)
 
     def get_workunits_internal(self) -> Iterable[MetadataWorkUnit]:
+
         for feature_view in self.feature_store.list_feature_views():
             for entity_name in feature_view.entities:
                 entity = self.feature_store.get_entity(entity_name)

metadata-ingestion/tests/integration/feast/feast_repository_mces_golden.json

@@ -11,6 +11,7 @@
                 },
                 {
                     "com.linkedin.pegasus2avro.ml.metadata.MLPrimaryKeyProperties": {
+                        "customProperties": {},
                         "description": "Driver ID",
                         "dataType": "ORDINAL",
                         "sources": [
@@ -23,7 +24,8 @@
     },
     "systemMetadata": {
         "lastObserved": 1586847600000,
-        "runId": "feast-repository-test"
+        "runId": "feast-repository-test",
+        "lastRunId": "no-run-id-provided"
     }
 },
 {
@@ -38,6 +40,7 @@
                 },
                 {
                     "com.linkedin.pegasus2avro.ml.metadata.MLFeatureProperties": {
+                        "customProperties": {},
                         "description": "Conv rate",
                         "dataType": "CONTINUOUS",
                         "sources": [
@@ -50,7 +53,8 @@
     },
     "systemMetadata": {
         "lastObserved": 1586847600000,
-        "runId": "feast-repository-test"
+        "runId": "feast-repository-test",
+        "lastRunId": "no-run-id-provided"
     }
 },
 {
@@ -65,6 +69,7 @@
                 },
                 {
                     "com.linkedin.pegasus2avro.ml.metadata.MLFeatureProperties": {
+                        "customProperties": {},
                         "description": "Acc rate",
                         "dataType": "CONTINUOUS",
                         "sources": [
@@ -77,7 +82,8 @@
     },
     "systemMetadata": {
         "lastObserved": 1586847600000,
-        "runId": "feast-repository-test"
+        "runId": "feast-repository-test",
+        "lastRunId": "no-run-id-provided"
     }
 },
 {
@@ -92,6 +98,7 @@
                 },
                 {
                     "com.linkedin.pegasus2avro.ml.metadata.MLFeatureProperties": {
+                        "customProperties": {},
                         "description": "Avg daily trips",
                         "dataType": "ORDINAL",
                         "sources": [
@@ -104,7 +111,8 @@
     },
     "systemMetadata": {
         "lastObserved": 1586847600000,
-        "runId": "feast-repository-test"
+        "runId": "feast-repository-test",
+        "lastRunId": "no-run-id-provided"
     }
 },
 {
@@ -119,6 +127,7 @@
                 },
                 {
                     "com.linkedin.pegasus2avro.ml.metadata.MLFeatureProperties": {
+                        "customProperties": {},
                         "description": "String feature",
                         "dataType": "TEXT",
                         "sources": [
@@ -131,7 +140,8 @@
     },
     "systemMetadata": {
         "lastObserved": 1586847600000,
-        "runId": "feast-repository-test"
+        "runId": "feast-repository-test",
+        "lastRunId": "no-run-id-provided"
     }
 },
 {
@@ -170,7 +180,8 @@
     },
     "systemMetadata": {
         "lastObserved": 1586847600000,
-        "runId": "feast-repository-test"
+        "runId": "feast-repository-test",
+        "lastRunId": "no-run-id-provided"
     }
 },
 {
@@ -189,7 +200,8 @@
     },
     "systemMetadata": {
         "lastObserved": 1586847600000,
-        "runId": "feast-repository-test"
+        "runId": "feast-repository-test",
+        "lastRunId": "no-run-id-provided"
     }
 },
 {
@@ -204,6 +216,7 @@
                 },
                 {
                     "com.linkedin.pegasus2avro.ml.metadata.MLFeatureProperties": {
+                        "customProperties": {},
                         "dataType": "CONTINUOUS",
                         "sources": [
                             "urn:li:dataset:(urn:li:dataPlatform:request,vals_to_add,PROD)",
@@ -216,7 +229,8 @@
     },
     "systemMetadata": {
         "lastObserved": 1586847600000,
-        "runId": "feast-repository-test"
+        "runId": "feast-repository-test",
+        "lastRunId": "no-run-id-provided"
     }
 },
 {
@@ -231,6 +245,7 @@
                 },
                 {
                     "com.linkedin.pegasus2avro.ml.metadata.MLFeatureProperties": {
+                        "customProperties": {},
                         "dataType": "CONTINUOUS",
                         "sources": [
                             "urn:li:dataset:(urn:li:dataPlatform:request,vals_to_add,PROD)",
@@ -243,7 +258,8 @@
     },
     "systemMetadata": {
         "lastObserved": 1586847600000,
-        "runId": "feast-repository-test"
+        "runId": "feast-repository-test",
+        "lastRunId": "no-run-id-provided"
     }
 },
 {
@@ -278,7 +294,8 @@
     },
     "systemMetadata": {
         "lastObserved": 1586847600000,
-        "runId": "feast-repository-test"
+        "runId": "feast-repository-test",
+        "lastRunId": "no-run-id-provided"
     }
 },
 {
@@ -297,7 +314,8 @@
     },
     "systemMetadata": {
         "lastObserved": 1586847600000,
-        "runId": "feast-repository-test"
+        "runId": "feast-repository-test",
+        "lastRunId": "no-run-id-provided"
     }
 }
 ]

metadata-ingestion/tests/integration/feast/feature_store/features.py

@@ -19,6 +19,8 @@
     join_keys=["driver_id"],
     value_type=ValueType.INT64,
     description="Driver ID",
+    owner="Datahub",
+    tags={"name": "deprecated"},
 )
 
 driver_hourly_stats_view = FeatureView(
@@ -29,7 +31,7 @@
         Field(
             name="conv_rate",
             dtype=feast.types.Float64,
-            tags=dict(description="Conv rate"),
+            tags={"name": "needs_documentation", "description": "Conv rate"},
         ),
         Field(
             name="acc_rate",
@@ -49,7 +51,8 @@
     ],
     online=True,
     source=driver_hourly_stats_source,
-    tags={},
+    tags={"name": "deprecated"},
+    owner="Datahub",
 )
 
 input_request = RequestSource(