Skip to content

Commit 9340545

Browse files
Marcus-RostiMarcus-Rosti
authored
feat: Add quality monitoring features for Trino offline store (#6779)
* feat: Add quality monitoring features for Trino offline store Signed-off-by: Marcus Rosti <[email protected]> * fix: Query structure and null counts Signed-off-by: Marcus Rosti <[email protected]> * fix: Infers null syntax from pandas Signed-off-by: Marcus Rosti <[email protected]> * fix: Guard _is_nan against container types in upload formatting Signed-off-by: Marcus Rosti <[email protected]> * fix: Implement rename-swap pattern for Trino monitoring tables Signed-off-by: Marcus Rosti <[email protected]> * fix: Use append-only writes for Trino monitoring metrics Signed-off-by: Marcus Rosti <[email protected]> * fix: Narrow exception handling and add debug logging in Trino monitoring Signed-off-by: Marcus Rosti <[email protected]> * fix: Convert tz-aware datetimes to UTC in Trino SQL literal formatting Signed-off-by: Marcus Rosti <[email protected]> --------- Signed-off-by: Marcus Rosti <[email protected]>
1 parent 5fea698 commit 9340545

6 files changed

Lines changed: 2021 additions & 28 deletions

File tree

‎docs/how-to-guides/feature-monitoring.md‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -359,6 +359,7 @@ Monitoring works natively with all offline stores that serve as compute engines
359359
| BigQuery | SQL push-down | `MERGE` into BQ tables |
360360
| Redshift | SQL push-down | `MERGE` via Data API |
361361
| Spark | SparkSQL push-down | Parquet tables |
362+
| Trino | SQL push-down | Trino tables |
362363
| Oracle | SQL via Ibis | `MERGE` from `DUAL` |
363364
| DuckDB | In-memory SQL | Parquet files |
364365
| Dask | PyArrow compute | Parquet files |

‎sdk/python/feast/infra/offline_stores/contrib/spark_offline_store/spark.py‎

Lines changed: 19 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -938,29 +938,33 @@ def _spark_sql_categorical_stats(
938938
) -> Dict[str, Any]:
939939
q_col = f"`{col_name}`"
940940

941-
query = (
942-
f"WITH filtered AS ("
943-
f" SELECT * FROM {from_expression} AS _src WHERE {ts_clause}"
944-
f") "
941+
counts_query = (
945942
f"SELECT "
946-
f" (SELECT COUNT(*) FROM filtered) AS row_count, "
947-
f" (SELECT COUNT(*) - COUNT({q_col}) FROM filtered) AS null_count, "
948-
f" (SELECT COUNT(DISTINCT {q_col}) FROM filtered "
949-
f" WHERE {q_col} IS NOT NULL) AS unique_count, "
950-
f" CAST({q_col} AS STRING) AS value, COUNT(*) AS cnt "
951-
f"FROM filtered WHERE {q_col} IS NOT NULL "
952-
f"GROUP BY {q_col} ORDER BY cnt DESC LIMIT {int(top_n)}"
943+
f" COUNT(*) AS row_count, "
944+
f" COUNT(*) - COUNT({q_col}) AS null_count, "
945+
f" COUNT(DISTINCT {q_col}) AS unique_count "
946+
f"FROM {from_expression} AS _src "
947+
f"WHERE {ts_clause}"
953948
)
954-
955-
rows = spark_session.sql(query).collect()
956-
if not rows:
949+
rows = spark_session.sql(counts_query).collect()
950+
if not rows or rows[0][0] is None or int(rows[0][0]) == 0:
957951
return empty_categorical_metric(col_name)
958952

959953
row_count = int(rows[0][0] or 0)
960954
null_count = int(rows[0][1] or 0)
961955
unique_count = int(rows[0][2] or 0)
962956

963-
top_entries = [{"value": r[3], "count": int(r[4] or 0)} for r in rows]
957+
top_entries: List[Dict[str, Any]] = []
958+
if row_count > null_count:
959+
top_n_query = (
960+
f"SELECT CAST({q_col} AS STRING) AS value, COUNT(*) AS cnt "
961+
f"FROM {from_expression} AS _src "
962+
f"WHERE {q_col} IS NOT NULL AND {ts_clause} "
963+
f"GROUP BY {q_col} ORDER BY cnt DESC LIMIT {int(top_n)}"
964+
)
965+
top_rows = spark_session.sql(top_n_query).collect()
966+
top_entries = [{"value": r[0], "count": int(r[1] or 0)} for r in top_rows]
967+
964968
top_total = sum(e["count"] for e in top_entries)
965969
other_count = (row_count - null_count) - top_total
966970

‎sdk/python/feast/infra/offline_stores/contrib/trino_offline_store/connectors/upload.py‎

Lines changed: 19 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,7 @@
1818
```
1919
"""
2020

21-
from datetime import datetime, timezone
21+
from datetime import date, datetime, timezone
2222
from typing import Any, Dict, Iterator, Optional, Set
2323

2424
import numpy as np
@@ -108,29 +108,38 @@ def format_pandas_row(df: pd.DataFrame) -> str:
108108
def _is_nan(value: Any) -> bool:
109109
if value is None:
110110
return True
111+
if isinstance(value, (list, tuple, np.ndarray)):
112+
return False
111113

112114
try:
113-
return np.isnan(value)
114-
except TypeError:
115+
return bool(pd.isna(value))
116+
except (TypeError, ValueError):
115117
return False
116118

117119
def _format_value(row: pd.Series, schema: Dict[str, Any]) -> str:
118120
formated_values = []
119121
for row_name, row_value in row.items():
120-
if schema[row_name].startswith("timestamp"):
121-
if isinstance(row_value, datetime):
122-
row_value = format_datetime(row_value)
123-
formated_values.append(f"TIMESTAMP '{row_value}'")
124-
elif isinstance(row_value, list):
122+
if isinstance(row_value, list):
125123
formated_values.append(f"ARRAY{row_value}")
126124
elif isinstance(row_value, np.ndarray):
127125
formated_values.append(f"ARRAY{row_value.tolist()}")
128126
elif isinstance(row_value, tuple):
129127
formated_values.append(f"ARRAY{list(row_value)}")
130-
elif isinstance(row_value, str):
131-
formated_values.append(f"'{row_value}'")
132128
elif _is_nan(row_value):
133129
formated_values.append("NULL")
130+
elif schema[row_name].startswith("timestamp"):
131+
if isinstance(row_value, datetime):
132+
row_value = format_datetime(row_value)
133+
formated_values.append(f"TIMESTAMP '{row_value}'")
134+
elif schema[row_name].startswith("date"):
135+
if isinstance(row_value, (datetime, date)):
136+
row_value = row_value.strftime("%Y-%m-%d")
137+
formated_values.append(f"DATE '{row_value}'")
138+
elif isinstance(row_value, (bool, np.bool_)):
139+
formated_values.append("TRUE" if row_value else "FALSE")
140+
elif isinstance(row_value, str):
141+
escaped = row_value.replace("'", "''")
142+
formated_values.append(f"'{escaped}'")
134143
else:
135144
formated_values.append(f"{row_value}")
136145

0 commit comments

Comments
 (0)