René's URL Explorer Experiment

{"@context":"https://schema.org","@type":"DiscussionForumPosting","headline":"NotImplementedError raised when using FeatureStore.push()","articleBody":"## Expected Behavior \r\nI'm currently starting to try simple usecase of Feast v0.40 for local development, I'm ingesting data from Airflow job to Postgres.\r\n```\r\ndef process():\r\n        ...\r\n        store = FeatureStore(repo_path=f\"{mother_path}/helpers/feast/\")\r\n\r\n        driver = Entity(name=\"driver_id\", join_keys=[\"driver_id\"], description=\"driver ID\")\r\n\r\n        table_name = 'feast_driver_hourly_stats'\r\n        create_postgres_table(\r\n            table_name=table_name, \r\n            schema=\"\"\"\r\n                driver_id SERIAL PRIMARY KEY,\r\n                event_timestamp TIMESTAMP,\r\n                surge_score FLOAT,\r\n                trip_cancel INT,\r\n                number_trip INT,\r\n                latest_completed INT\r\n            \"\"\"\r\n        )\r\n        \r\n\r\n        driver_stats_source = PostgreSQLSource(\r\n            name=table_name,\r\n            query=f\"SELECT * FROM airflow.feast.{table_name}\",\r\n            timestamp_field=\"event_timestamp\",\r\n        )\r\n\r\n        push_source = PushSource(\r\n            name=\"push_source\",\r\n            batch_source=driver_stats_source\r\n        )\r\n\r\n        # Define the feature view (without specifying a source since we'll use a DataFrame)\r\n        driver_feature_view = FeatureView(\r\n            name=\"driver_feature_view\",\r\n            entities=[driver],\r\n            ttl=None,  # Set TTL if needed\r\n            schema=[\r\n                Field(name=\"surge_score\", dtype=Float32),\r\n                Field(name=\"trip_cancel\", dtype=Int64),\r\n                Field(name=\"number_trip\", dtype=Int64),\r\n                Field(name=\"latest_completed\", dtype=Int64),\r\n            ],\r\n            # online=True,\r\n            stream_source=push_source\r\n        )\r\n\r\n        # Apply the feature store (register entities and feature views)\r\n        store.apply([driver_feature_view])\r\n        result = store.push(\"push_source\", data, to=PushMode.OFFLINE)\r\n```\r\nI have ensured that the database, the schema, and the table exists. I expect after calling push(), the data will then be ingested (written) into the destined table.\r\n\r\n## Current Behavior\r\nSuddenly, I got the follow error:\r\n```\r\nFile \"/opt/airflow/dags/test_feast.py\", line 112, in process\r\n    result = store.push(\"push_source\", data, to=PushMode.OFFLINE)\r\n             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\r\n  File \"/home/airflow/.local/lib/python3.11/site-packages/feast/feature_store.py\", line 1427, in push\r\n    self.write_to_offline_store(\r\n  File \"/home/airflow/.local/lib/python3.11/site-packages/feast/feature_store.py\", line 1511, in write_to_offline_store\r\n    provider.ingest_df_to_offline_store(feature_view, table)\r\n  File \"/home/airflow/.local/lib/python3.11/site-packages/feast/infra/passthrough_provider.py\", line 303, in ingest_df_to_offline_store\r\n    self.offline_write_batch(self.repo_config, feature_view, table, None)\r\n  File \"/home/airflow/.local/lib/python3.11/site-packages/feast/infra/passthrough_provider.py\", line 180, in offline_write_batch\r\n    self.offline_store.__class__.offline_write_batch(\r\n  File \"/home/airflow/.local/lib/python3.11/site-packages/feast/infra/offline_stores/offline_store.py\", line 353, in offline_write_batch\r\n    raise NotImplementedError\r\n```\r\nThe `write_to_offline_store()` method might have called directly to an interface, causing the bug. I'm currently trying to review my configs/codes to spot any mistake and also read the `push()` method. However, I think I might have to raise an issue here.\r\n\r\n### Specifications\r\n\r\n- Version: v0.40\r\n- Platform: Airflow, Postgres\r\n- Subsystem: Python 3.11\r\n","author":{"url":"https://github.com/singsinghai","@type":"Person","name":"singsinghai"},"datePublished":"2024-08-13T10:52:02.000Z","interactionStatistic":{"@type":"InteractionCounter","interactionType":"https://schema.org/CommentAction","userInteractionCount":4},"url":"https://github.com/4403/feast/issues/4403"}