Custom data sources fail for PushSource

[mmarqz]

Expected Behavior

Custom data sources should work with PushSource when running test_workflow.py

driver_stats_push_source = PushSource(
    name="driver_stats_push_source",
    batch_source=custom_driver_stats_source,
)

Current Behavior

Running test_workflow.py throws the following exception when calling get_historical_features if a custom data source was defined for a PushSource as above:

python3.9/site-packages/feast/repo_config.py", line 486, in get_data_source_class_from_type
    module_name, config_class_name = data_source_type.rsplit(".", 1)
ValueError: not enough values to unpack (expected 2, got 1)

Steps to reproduce

• run feast init
• define new custom data source as defined in docs:

class CustomFileDataSource(FileSource):
    """Custom data source class for local files"""
    def __init__(
        self,
        event_timestamp_column: Optional[str] = "",
        path: Optional[str] = None,
        field_mapping: Optional[Dict[str, str]] = None,
        created_timestamp_column: Optional[str] = "",
        date_partition_column: Optional[str] = "",
    ):
        super(CustomFileDataSource, self).__init__(
            event_timestamp_column=event_timestamp_column,
            created_timestamp_column=created_timestamp_column,
            field_mapping=field_mapping,
            date_partition_column=date_partition_column,
            path=path,
        )
        self._path = path

    @property
    def path(self):
        """
        Returns the file path of this feature data source.
        """
        return self._path

    @staticmethod
    def from_proto(data_source: DataSourceProto):
        """
        Creates a `CustomFileDataSource` object from a DataSource proto, by
        parsing the CustomSourceOptions which is encoded as a binary json string.
        """
        custom_source_options = str(
            data_source.custom_options.configuration, encoding="utf8"
        )
        path = json.loads(custom_source_options)["path"]
        return CustomFileDataSource(
            field_mapping=dict(data_source.field_mapping),
            path=path,
            event_timestamp_column=data_source.event_timestamp_column,
            created_timestamp_column=data_source.created_timestamp_column,
            date_partition_column=data_source.date_partition_column,
        )

    def to_proto(self) -> DataSourceProto:
        """
        Creates a DataSource proto representation of this object, by serializing some
        custom options into the custom_options field as a binary encoded json string.
        """
        config_json = json.dumps({"path": self.path})
        data_source_proto = DataSourceProto(
            type=DataSourceProto.CUSTOM_SOURCE,
            field_mapping=self.field_mapping,
            custom_options=DataSourceProto.CustomSourceOptions(
                configuration=bytes(config_json, encoding="utf8")
            ),
        )

        data_source_proto.event_timestamp_column = self.event_timestamp_column
        data_source_proto.created_timestamp_column = self.created_timestamp_column
        data_source_proto.date_partition_column = self.date_partition_column

        return data_source_proto

    def get_table_query_string(self) -> str:
        pass

• Modify PushSource to use new CustomFileDataSource as batch_source
• Run test_workflows.py

Specifications

• Version: 0.31.1
• Platform: Mac
• Subsystem:

Possible Solution

[npsables]

Hi, did u resolve this?

[stale]

This issue has been automatically marked as stale because it has not had recent activity. It will be closed if no further activity occurs. Thank you for your contributions.