WeichenXu123 commented on code in PR #37734: URL: https://github.com/apache/spark/pull/37734#discussion_r1016162543
########## python/pyspark/ml/functions.py: ########## @@ -106,6 +117,556 @@ def array_to_vector(col: Column) -> Column: return Column(sc._jvm.org.apache.spark.ml.functions.array_to_vector(_to_java_column(col))) +def _batched( + data: pd.Series | pd.DataFrame | Tuple[pd.Series], batch_size: int +) -> Iterator[pd.DataFrame]: + """Generator that splits a pandas dataframe/series into batches.""" + if isinstance(data, pd.DataFrame): + df = data + elif isinstance(data, pd.Series): + df = pd.concat((data,), axis=1) + else: # isinstance(data, Tuple[pd.Series]): + df = pd.concat(data, axis=1) + + index = 0 + data_size = len(df) + while index < data_size: + yield df.iloc[index : index + batch_size] + index += batch_size + + +def _is_tensor_col(data: pd.Series | pd.DataFrame) -> bool: + if isinstance(data, pd.Series): + return data.dtype == np.object_ and isinstance(data.iloc[0], (np.ndarray, list)) + elif isinstance(data, pd.DataFrame): + return any(data.dtypes == np.object_) and any( + [isinstance(d, (np.ndarray, list)) for d in data.iloc[0]] + ) + else: + raise ValueError( + "Unexpected data type: {}, expected pd.Series or pd.DataFrame.".format(type(data)) + ) + + +def _has_tensor_cols(data: pd.Series | pd.DataFrame | Tuple[pd.Series]) -> bool: + """Check if input Series/DataFrame/Tuple contains any tensor-valued columns.""" + if isinstance(data, (pd.Series, pd.DataFrame)): + return _is_tensor_col(data) + else: # isinstance(data, Tuple): + return any(_is_tensor_col(elem) for elem in data) + + +def _validate_and_transform_prediction_result( + preds: np.ndarray | Mapping[str, np.ndarray] | List[Mapping[str, Any]], + num_input_rows: int, + return_type: DataType, +) -> pd.DataFrame | pd.Series: + """Validate numpy-based model predictions against the expected pandas_udf return_type and + transforms the predictions into an equivalent pandas DataFrame or Series.""" + if isinstance(return_type, StructType): + struct_rtype: StructType = return_type + fieldNames = struct_rtype.names + if isinstance(preds, dict): + # dictionary of columns + predNames = list(preds.keys()) + for field in struct_rtype.fields: + if isinstance(field.dataType, ArrayType): Review Comment: Add check: for other cases (non-ArrayType), pls check that the value shape is 1 dimension, or 2-dimension array but the second dim = 1. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org --------------------------------------------------------------------- To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org