Method write_to_online_store_async

sdk/python/feast/feature_store.py:3128–3169

Persists a dataframe to the online store asynchronously. Args: feature_view_name: The feature view to which the dataframe corresponds. df: The dataframe to be persisted. inputs: Optional the dictionary object to be written allow_regis

(
        self,
        feature_view_name: str,
        df: Optional[pd.DataFrame] = None,
        inputs: Optional[Union[Dict[str, List[Any]], pd.DataFrame]] = None,
        allow_registry_cache: bool = True,
    )

        provider.ingest_df(feature_view, df)

    async def write_to_online_store_async(
        self,
        feature_view_name: str,
        df: Optional[pd.DataFrame] = None,
        inputs: Optional[Union[Dict[str, List[Any]], pd.DataFrame]] = None,
        allow_registry_cache: bool = True,
    ):
        """
        Persists a dataframe to the online store asynchronously.

        Args:
            feature_view_name: The feature view to which the dataframe corresponds.
            df: The dataframe to be persisted.
            inputs: Optional the dictionary object to be written
            allow_registry_cache (optional): Whether to allow retrieving feature views from a cached registry.
        """

        feature_view, df = self._get_feature_view_and_df_for_online_write(
            feature_view_name=feature_view_name,
            df=df,
            inputs=inputs,
            allow_registry_cache=allow_registry_cache,
        )

        # Validate that the dataframe has meaningful feature data
        if df is not None:
            if df.empty:
                warnings.warn("Cannot write empty dataframe to online store")
                return  # Early return for empty dataframe

            # Check if feature columns are empty (entity columns may have data but feature columns are empty)
            feature_column_names = [f.name for f in feature_view.features]
            if feature_column_names:
                feature_df = df[feature_column_names]
                if feature_df.empty or feature_df.isnull().all().all():
                    warnings.warn(
                        "Cannot write dataframe with empty feature columns to online store"
                    )
                    return  # Early return for empty feature columns

        provider = self._get_provider()
        await provider.ingest_df_async(feature_view, df)

    async def update_online_store(
        self,