Feast on Amazon 解决方案

project: feast_spark_project
registry: data/registry.db
provider: local
offline_store:
    type: spark
    spark_conf:
        spark.master: yarn
        spark.ui.enabled: "true"
        spark.eventLog.enabled: "true"
        spark.sql.catalogImplementation: "hive"
        spark.sql.parser.quotedRegexColumnNames: "true"
        spark.sql.session.timeZone: "UTC"

source /etc/spark/conf/spark-env.sh
export PYTHONPATH="${SPARK_HOME}/python/:$PYTHONPATH"
export PYTHONPATH="${SPARK_HOME}/python/lib/py4j-0.10.9-src.zip:$PYTHONPATH"

from feast.infra.offline_stores.contrib.spark_offline_store.spark_source import ( SparkSource,)

driver_hourly_stats= SparkSource(
        name="driver_hourly_stats",
        query="SELECT event_timestamp as ts, created_timestamp as  created, conv_rate,conv_rate,conv_rate FROM emr_feature_store.driver_hourly_stats",
        event_timestamp_column="ts",
        created_timestamp_column="created"
        )

== Parsed Logical Plan ==
'CreateViewStatement [driver_hourly_stats__cleaned], (

    WITH driver_hourly_stats__entity_dataframe AS (
        SELECT
            driver_id,
            entity_timestamp,
            driver_hourly_stats__entity_row_unique_id
        FROM entity_dataframe
        GROUP BY
            driver_id,
            entity_timestamp,
            driver_hourly_stats__entity_row_unique_id
    ),

driver_hourly_stats__base AS (
        SELECT
            subquery.*,
            entity_dataframe.entity_timestamp,
            entity_dataframe.driver_hourly_stats__entity_row_unique_id
        FROM driver_hourly_stats__subquery AS subquery
        INNER JOIN driver_hourly_stats__entity_dataframe AS entity_dataframe
        ON TRUE
            AND subquery.event_timestamp <= entity_dataframe.entity_timestamp

            
            AND subquery.event_timestamp >= entity_dataframe.entity_timestamp - 86400 * interval '1' second
            

            
            AND subquery.driver_id = entity_dataframe.driver_id
            
    ),

driver_hourly_stats__subquery AS (
        SELECT
            ts as event_timestamp,
            created as created_timestamp,
            driver_id AS driver_id,
            
                conv_rate as conv_rate, 
            
                acc_rate as acc_rate
            
        FROM (SELECT driver_id,event_timestamp as ts, created_timestamp as  created, conv_rate,acc_rate,avg_daily_trips FROM emr_feature_store.driver_hourly_stats)
        WHERE ts <= '2022-07-25T03:27:05.903000'
        
        AND ts >= '2022-07-21T03:27:05.903000'
        
    ),

 driver_hourly_stats__dedup AS (
        SELECT
            driver_hourly_stats__entity_row_unique_id,
            event_timestamp,
            MAX(created_timestamp) as created_timestamp
        FROM driver_hourly_stats__base
        GROUP BY driver_hourly_stats__entity_row_unique_id, event_timestamp
    ),
driver_hourly_stats__latest AS (
        SELECT
            event_timestamp,
            created_timestamp,
            driver_hourly_stats__entity_row_unique_id
        FROM
        (
            SELECT *,
                ROW_NUMBER() OVER(
                    PARTITION BY driver_hourly_stats__entity_row_unique_id
                    ORDER BY event_timestamp DESC,created_timestamp DESC
                ) AS row_number
            FROM driver_hourly_stats__base
            
                INNER JOIN driver_hourly_stats__dedup
                USING (driver_hourly_stats__entity_row_unique_id, event_timestamp, created_timestamp)
            
        )
        WHERE row_number = 1
    )