fix: Multiple Spark Enhancements (#1800)

* fix: update spark numeric stats calculations

In the Pandas implementation, the numeric stats like min/max/stddev/etc.
by default ignore null values.
This commit updates the spark implementation to more closely match that.

* fix: update spark null checks for describe_counts_spark method

Need to add the isnan() check because Pandas isnull check will count NaN as null,
 but Spark does not.

* fix: update Spark frequency counts

The previous calculation of counts was actually counting an already summarized
dataframe, so it wasn't capturing the correct counts for each instance of a value.

This is updated by summing the count value instead of performing a row count.

* fix: adding tests for issue 1429

* fix: edge case - completely null numeric field in Spark

Discovered this edge case with real data, and still need to fix the rendering of an
empty histogram.

* fix: add handling for spark DecimalType

This change addresses issue #1602.

Computations in the summarize process result in some floats when computing against
decimal columns.To solution this, we simply convert those types to a DoubleType
when performing those numeric operations.

* fix: add handling for spark correlations with no numeric fields

Assembling a vector column in Spark with no numeric columns results in features with
a NULL size, NULL indices, and an empty list of values.
This causes an exception to be raised when computing correlations.

The solution here is to avoid computing the correlation matrix when there are no
interval columns (numeric).

This change addresses issue #1722.

* fix: allow NoneType values to be string formatted

This change addresses issue #1723.
It implements a "N/A" string as the default when formatting NoneType values.

* fix: multiple Spark fixes to approach closer parity to Pandas profiles

Addresses handling of completely null numeric columns, and gracefully handling
empty correlation sets and plots.

Author: MCBoarder289

.gitignore

@@ -24,6 +24,7 @@ var/
 *.egg-info/
 .installed.cfg
 *.egg
+uv.lock
 
 # PyInstaller
 #  Usually these files are written by a python script from a template
@@ -85,3 +86,6 @@ docsrc/source/pages/api/_autosummary/
 # User created
 VERSION
 version.py
+
+# local java/spark sdk environment files
+.sdkmanrc

pyproject.toml

@@ -87,6 +87,7 @@ dev = [
     "sphinx-autodoc-typehints>=1.10.3",
     "sphinx-multiversion>=0.2.3",
     "autodoc_pydantic",
+    "standard-imghdr"
 ]
 
 docs = [

src/ydata_profiling/model/spark/correlations_spark.py

@@ -45,10 +45,23 @@ def _compute_corr_natively(df: DataFrame, summary: dict, corr_type: str) -> Arra
     exact same workflow. The syntax is Correlation.corr(dataframe, method="pearson" OR "spearman"),
     and Correlation is from pyspark.ml.stat
     """
-    variables = {column: description["type"] for column, description in summary.items()}
+    variables = {
+        column: {
+            "type": description["type"],
+            "all_missing": description.get("n", 0) - description.get("n_missing", 0)
+            == 0,
+        }
+        for column, description in summary.items()
+    }
     interval_columns = [
-        column for column, type_name in variables.items() if type_name == "Numeric"
+        column
+        for column, mapping in variables.items()
+        if mapping["type"] == "Numeric" and not mapping["all_missing"]
     ]
+
+    if not interval_columns:
+        return [], interval_columns
+
     df = df.select(*interval_columns)
 
     # convert to vector column first
@@ -63,6 +76,9 @@ def _compute_corr_natively(df: DataFrame, summary: dict, corr_type: str) -> Arra
     assembler = VectorAssembler(**assembler_args)
     df_vector = assembler.transform(df).select(vector_col)
 
+    if df_vector.count() <= 1:
+        return [], interval_columns
+
     # get correlation matrix
     matrix = (
         Correlation.corr(df_vector, vector_col, method=corr_type).head()[0].toArray()

src/ydata_profiling/model/spark/describe_counts_spark.py

@@ -6,6 +6,7 @@
 import pandas as pd
 from pyspark.sql import DataFrame
 from pyspark.sql import functions as F
+from pyspark.sql import types as T
 
 from ydata_profiling.config import Settings
 from ydata_profiling.model.summary_algorithms import describe_counts
@@ -25,6 +26,11 @@ def describe_counts_spark(
     Returns:
         Updated settings, input series, and summary dictionary.
     """
+    # Cast Decimal Type s
+    if isinstance(series.schema.fields[0].dataType, T.DecimalType):
+        series = series.select(
+            F.col(series.columns[0]).cast(T.DoubleType()).alias(series.columns[0])
+        )
 
     # Count occurrences of each value
     value_counts = series.groupBy(series.columns[0]).count()
@@ -36,9 +42,23 @@ def describe_counts_spark(
     value_counts_index_sorted = value_counts.orderBy(F.asc(series.columns[0]))
 
     # Count missing values
-    n_missing = (
-        value_counts.filter(F.col(series.columns[0]).isNull()).select("count").first()
-    )
+    if series.dtypes[0][1] in ("int", "float", "bigint", "double"):
+        n_missing = (
+            # Need to add the isnan() check because Pandas isnull check will count NaN as null, but Spark does not
+            value_counts.filter(
+                F.col(series.columns[0]).isNull() | F.isnan(F.col(series.columns[0]))
+            )
+            .select("count")
+            .first()
+        )
+    else:
+        n_missing = (
+            # Need to add the isnan() check because Pandas isnull check will count NaN as null, but Spark does not
+            value_counts.filter(F.col(series.columns[0]).isNull())
+            .select("count")
+            .first()
+        )
+
     n_missing = n_missing["count"] if n_missing else 0
 
     # Convert top 200 values to Pandas for frequency table display
@@ -60,17 +80,15 @@ def describe_counts_spark(
             value_counts.filter(F.col(column).isNotNull())  # Exclude NaNs
             .filter(~F.isnan(F.col(column)))  # Remove implicit NaNs (if numeric column)
             .groupBy(column)  # Group by unique values
-            .count()  # Count occurrences
+            .agg(F.sum("count").alias("count"))  # Sum of count
             .orderBy(F.desc("count"))  # Sort in descending order
-            .limit(200)  # Limit for performance
         )
     else:
         value_counts_no_nan = (
             value_counts.filter(F.col(column).isNotNull())  # Exclude NULLs
             .groupBy(column)  # Group by unique timestamp values
-            .count()  # Count occurrences
+            .agg(F.sum("count").alias("count"))  # Sum of count
             .orderBy(F.desc("count"))  # Sort by most frequent timestamps
-            .limit(200)  # Limit for performance
         )
 
     # Convert to Pandas Series, forcing proper structure

src/ydata_profiling/model/spark/describe_numeric_spark.py

@@ -11,6 +11,14 @@
 def numeric_stats_spark(df: DataFrame, summary: dict) -> dict:
     column = df.columns[0]
 
+    # Removing null types from numeric summary stats to match Pandas defaults which skip na's (skipna=False)
+    finite_filter = (
+        F.col(column).isNotNull()
+        & ~F.isnan(F.col(column))
+        & ~F.col(column).isin([np.inf, -np.inf])
+    )
+    non_null_df = df.filter(finite_filter)
+
     expr = [
         F.mean(F.col(column)).alias("mean"),
         F.stddev(F.col(column)).alias("std"),
@@ -21,7 +29,7 @@ def numeric_stats_spark(df: DataFrame, summary: dict) -> dict:
         F.skewness(F.col(column)).alias("skewness"),
         F.sum(F.col(column)).alias("sum"),
     ]
-    return df.agg(*expr).first().asDict()
+    return non_null_df.agg(*expr).first().asDict()
 
 
 def describe_numeric_1d_spark(
@@ -81,30 +89,42 @@ def describe_numeric_1d_spark(
     quantiles = config.vars.num.quantiles
     quantile_threshold = 0.05
 
-    summary.update(
-        {
-            f"{percentile:.0%}": value
-            for percentile, value in zip(
-                quantiles,
-                df.stat.approxQuantile(
-                    f"{df.columns[0]}",
+    if summary.get("n") == summary.get("n_missing"):
+        # This means the entire column is null/nan, so summary values need to be hard-coded:
+        summary.update({f"{percentile:.0%}": np.nan for percentile in quantiles})
+
+        summary["mad"] = np.nan
+        summary["iqr"] = np.nan
+
+    else:
+        summary.update(
+            {
+                f"{percentile:.0%}": value
+                for percentile, value in zip(
                     quantiles,
-                    quantile_threshold,
-                ),
-            )
-        }
-    )
+                    df.stat.approxQuantile(
+                        f"{df.columns[0]}",
+                        quantiles,
+                        quantile_threshold,
+                    ),
+                )
+            }
+        )
 
-    median = summary["50%"]
+        median = summary.get("50%")
 
-    summary["mad"] = df.select(
-        (F.abs(F.col(f"{df.columns[0]}").cast("int") - median)).alias("abs_dev")
-    ).stat.approxQuantile("abs_dev", [0.5], quantile_threshold)[0]
+        summary["mad"] = df.select(
+            (F.abs(F.col(f"{df.columns[0]}").cast("int") - median)).alias("abs_dev")
+        ).stat.approxQuantile("abs_dev", [0.5], quantile_threshold)[0]
+
+        summary["iqr"] = summary["75%"] - summary["25%"]
 
     # FIXME: move to fmt
     summary["p_negative"] = summary["n_negative"] / summary["n"]
-    summary["range"] = summary["max"] - summary["min"]
-    summary["iqr"] = summary["75%"] - summary["25%"]
+    if summary["min"] is None or summary["max"] is None:
+        summary["range"] = np.nan
+    else:
+        summary["range"] = summary["max"] - summary["min"]
     summary["cv"] = summary["std"] / summary["mean"] if summary["mean"] else np.nan
     summary["p_zeros"] = summary["n_zeros"] / summary["n"]
     summary["p_infinite"] = summary["n_infinite"] / summary["n"]

src/ydata_profiling/model/spark/describe_supported_spark.py

@@ -19,15 +19,15 @@ def describe_supported_spark(
     """
 
     # number of non-NaN observations in the Series
-    count = summary["count"]
-    n_distinct = summary["value_counts"].count()
+    n = summary["n"]
+    n_distinct = summary["value_counts"].count()  # Null/nan's count in value_counts
 
     summary["n_distinct"] = n_distinct
-    summary["p_distinct"] = n_distinct / count if count > 0 else 0
+    summary["p_distinct"] = n_distinct / n if n > 0 else 0
 
     n_unique = summary["value_counts"].where("count == 1").count()
-    summary["is_unique"] = n_unique == count
+    summary["is_unique"] = n_unique == n
     summary["n_unique"] = n_unique
-    summary["p_unique"] = n_unique / count if count > 0 else 0
+    summary["p_unique"] = n_unique / n if n > 0 else 0
 
     return config, series, summary

src/ydata_profiling/model/spark/summary_spark.py

@@ -43,6 +43,8 @@ def spark_describe_1d(
 
         if str(series.schema[0].dataType).startswith("ArrayType"):
             dtype = "ArrayType"
+        elif str(series.schema[0].dataType).startswith("Decimal"):
+            dtype = "decimal"
         else:
             dtype = series.schema[0].dataType.simpleString()
 
@@ -56,6 +58,7 @@ def spark_describe_1d(
             "boolean": "Boolean",
             "date": "DateTime",
             "timestamp": "DateTime",
+            "decimal": "Numeric",
         }[dtype]
 
     return summarizer.summarize(config, series, dtype=vtype)

src/ydata_profiling/report/formatters.py

@@ -245,13 +245,16 @@ def fmt_numeric(value: float, precision: int = 10) -> str:
     Returns:
         The numeric value with the given precision.
     """
-    fmtted = f"{{:.{precision}g}}".format(value)
-    for v in ["e+", "e-"]:
-        if v in fmtted:
-            sign = "-" if v in "e-" else ""
-            fmtted = fmtted.replace(v, " × 10<sup>") + "</sup>"
-            fmtted = fmtted.replace("<sup>0", "<sup>")
-            fmtted = fmtted.replace("<sup>", f"<sup>{sign}")
+    if value is None:
+        fmtted = "N/A"
+    else:
+        fmtted = f"{{:.{precision}g}}".format(value)
+        for v in ["e+", "e-"]:
+            if v in fmtted:
+                sign = "-" if v in "e-" else ""
+                fmtted = fmtted.replace(v, " × 10<sup>") + "</sup>"
+                fmtted = fmtted.replace("<sup>0", "<sup>")
+                fmtted = fmtted.replace("<sup>", f"<sup>{sign}")
 
     return fmtted
 

src/ydata_profiling/report/structure/variables/render_categorical.py

@@ -23,6 +23,7 @@
 from ydata_profiling.report.presentation.core.renderable import Renderable
 from ydata_profiling.report.presentation.frequency_table_utils import freq_table
 from ydata_profiling.report.structure.variables.render_common import render_common
+from ydata_profiling.report.utils import image_or_empty
 from ydata_profiling.visualisation.plot import cat_frequency_plot, histogram
 
 
@@ -95,7 +96,7 @@ def render_categorical_length(
     else:
         hist_data = histogram(config, *summary["histogram_length"])
 
-    length_histo = Image(
+    length_histo = image_or_empty(
         hist_data,
         image_format=config.plot.image_format,
         alt="length histogram",

src/ydata_profiling/report/structure/variables/render_count.py

@@ -8,11 +8,11 @@
 from ydata_profiling.report.presentation.core import (
     Container,
     FrequencyTable,
-    Image,
     Table,
     VariableInfo,
 )
 from ydata_profiling.report.structure.variables.render_common import render_common
+from ydata_profiling.report.utils import image_or_empty
 from ydata_profiling.visualisation.plot import histogram, mini_histogram
 
 
@@ -94,26 +94,41 @@ def render_count(config: Settings, summary: dict) -> dict:
         style=config.html.style,
     )
 
-    mini_histo = Image(
-        mini_histogram(config, *summary["histogram"]),
-        image_format=image_format,
+    summary_histogram = summary.get("histogram", [])
+
+    mini_hist_data = None
+
+    if summary_histogram:
+        mini_hist_data = mini_histogram(config, *summary["histogram"])
+
+    mini_histo = image_or_empty(
+        mini_hist_data,
         alt="Mini histogram",
+        image_format=image_format,
+        name="Mini Histogram",
     )
 
     template_variables["top"] = Container(
         [info, table1, table2, mini_histo], sequence_type="grid"
     )
 
-    seqs = [
-        Image(
-            histogram(config, *summary["histogram"]),
-            image_format=image_format,
-            alt="Histogram",
-            caption=f"<strong>Histogram with fixed size bins</strong> (bins={len(summary['histogram'][1]) - 1})",
-            name="Histogram",
-            anchor_id="histogram",
-        )
-    ]
+    hist_data = None
+    hist_caption = None
+
+    if summary_histogram:
+        hist_data = histogram(config, *summary["histogram"])
+        hist_caption = f"<strong>Histogram with fixed size bins</strong> (bins={len(summary['histogram'][1]) - 1})"
+
+    hist = image_or_empty(
+        hist_data,
+        alt="Histogram",
+        image_format=image_format,
+        caption=hist_caption,
+        name="Histogram",
+        anchor_id="histogram",
+    )
+
+    seqs = [hist]
 
     fq = FrequencyTable(
         template_variables["freq_table_rows"],