Class FeatureServiceLoggingSource

sdk/python/feast/feature_logging.py:45–105

class FeatureServiceLoggingSource(LoggingSource):
    def __init__(self, feature_service: "FeatureService", project: str):
        self._feature_service = feature_service
        self._project = project

    def get_schema(self, registry: "BaseRegistry") -> pa.Schema:
        fields: Dict[str, pa.DataType] = {}

        for projection in self._feature_service.feature_view_projections:
            # The order of fields in the generated schema should match
            # the order created on the other side (inside Go logger).
            # Otherwise, some offline stores might not accept parquet files (produced by Go).
            # Go code can be found here:
            # https://github.com/feast-dev/feast/blob/master/go/internal/feast/server/logging/memorybuffer.go#L51
            try:
                feast_object = get_feature_view_from_registry(
                    registry,
                    projection.name,
                    self._project,
                )
            except FeastObjectNotFoundException:
                raise FeastObjectNotFoundException(
                    f"Can't recognize feature view with a name {projection.name}"
                )

            if hasattr(feast_object, "source_request_sources"):
                for request_source in feast_object.source_request_sources.values():
                    for field in request_source.schema:
                        fields[field.name] = FEAST_TYPE_TO_ARROW_TYPE[field.dtype]
            else:
                for entity_column in feast_object.entity_columns:
                    if entity_column.name == DUMMY_ENTITY_ID:
                        continue

                    join_key = projection.join_key_map.get(
                        entity_column.name, entity_column.name
                    )
                    fields[join_key] = FEAST_TYPE_TO_ARROW_TYPE[entity_column.dtype]

            for feature in projection.features:
                fields[f"{projection.name_to_use()}__{feature.name}"] = (
                    FEAST_TYPE_TO_ARROW_TYPE[feature.dtype]
                )
                fields[f"{projection.name_to_use()}__{feature.name}__timestamp"] = (
                    PA_TIMESTAMP_TYPE
                )
                fields[f"{projection.name_to_use()}__{feature.name}__status"] = (
                    pa.int32()
                )

        # system columns
        fields[LOG_TIMESTAMP_FIELD] = pa.timestamp("us", tz=timezone.utc)
        fields[LOG_DATE_FIELD] = pa.date32()
        fields[REQUEST_ID_FIELD] = pa.string()

        return pa.schema(
            [pa.field(name, data_type) for name, data_type in fields.items()]
        )