add spark description dataclass

Author: vorel99

src/ydata_profiling/model/pandas/var_description/counts_pandas.py

@@ -67,4 +67,5 @@ def __init__(self, config: Settings, series: pd.Series):
             p_missing=series.isna().sum() / length if length > 0 else 0,
             count=length - series.isna().sum(),
             memory_size=series.memory_usage(deep=config.memory_deep),
+            value_counts=None,
         )

src/ydata_profiling/model/pandas/var_description/default_pandas.py

@@ -65,6 +65,7 @@ def from_var_counts(
             n_unique=unique_count,
             p_unique=unique_count / _count if _count > 0 else 0,
             var_specific=init_dict,
+            value_counts=var_counts.value_counts,
         )
 
 

src/ydata_profiling/model/spark/var_description/counts_spark.py

@@ -0,0 +1,72 @@
+from pyspark.sql import DataFrame
+
+from ydata_profiling.config import Settings
+from ydata_profiling.model.var_description.counts import VarCounts
+
+
+class VarCountsSpark(VarCounts):
+    value_counts_without_nan: DataFrame
+    """Counts of values in the series without NaN."""
+    value_counts_index_sorted: DataFrame
+    """Sorted counts of values in the series without NaN."""
+    value_counts: DataFrame
+
+    def __init__(self, config: Settings, series: DataFrame):
+        """Counts the values in a series (with and without NaN, distinct).
+
+        Args:
+            config: report Settings object
+            series: Series for which we want to calculate the values.
+            summary: series' summary
+
+        Returns:
+            A dictionary with the count values (with and without NaN, distinct).
+        """
+        length = series.count()
+
+        value_counts = series.groupBy(series.columns).count()
+        value_counts = value_counts.sort("count", ascending=False).persist()
+        value_counts_index_sorted = value_counts.sort(series.columns[0], ascending=True)
+
+        n_missing = value_counts.where(value_counts[series.columns[0]].isNull()).first()
+        if n_missing is None:
+            n_missing = 0
+        else:
+            n_missing = n_missing["count"]
+
+        # FIXME: reduce to top-n and bottom-n
+        value_counts_index_sorted = (
+            value_counts_index_sorted.limit(200)
+            .toPandas()
+            .set_index(series.columns[0], drop=True)
+            .squeeze(axis="columns")
+        )
+
+        # this is necessary as freqtables requires value_counts_without_nan
+        # to be a pandas series. However, if we try to get everything into
+        # pandas we will definitly crash the server
+        value_counts_without_nan = (
+            value_counts.dropna()
+            .limit(200)
+            .toPandas()
+            .set_index(series.columns[0], drop=True)
+            .squeeze(axis="columns")
+        )
+
+        # FIXME: This is not correct, but used to fulfil render expectations
+        # @chanedwin
+        memory_size = 0
+
+        self.value_counts = value_counts
+        super().__init__(
+            hashable=False,
+            value_counts_without_nan=value_counts_without_nan,
+            value_counts_index_sorted=value_counts_index_sorted,
+            ordering=False,
+            n_missing=n_missing,
+            n=length,
+            p_missing=n_missing / length,
+            count=length - n_missing,
+            memory_size=memory_size,
+            value_counts=value_counts.persist(),
+        )

src/ydata_profiling/model/spark/var_description/default_spark.py

@@ -0,0 +1,55 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from pyspark.sql import DataFrame
+
+from ydata_profiling.config import Settings
+from ydata_profiling.model.spark.var_description.counts_spark import VarCountsSpark
+from ydata_profiling.model.var_description.default import VarDescriptionHashable
+
+
+@dataclass
+class VarDescriptionSparkHashable(VarDescriptionHashable):
+    """Default description for pandas columns."""
+
+    @classmethod
+    def from_var_counts(
+        cls, var_counts: VarCountsSpark, init_dict: dict
+    ) -> VarDescriptionSparkHashable:
+        """Get a default description from a VarCountsPandas object."""
+
+        count = var_counts.count
+        n_distinct = var_counts.value_counts.count()
+
+        p_distinct = n_distinct / count if count > 0 else 0
+
+        n_unique = var_counts.value_counts.where("count == 1").count()
+        is_unique = n_unique == count
+        p_unique = n_unique / count
+
+        return VarDescriptionSparkHashable(
+            n=var_counts.n,
+            count=var_counts.count,
+            n_missing=var_counts.n_missing,
+            p_missing=var_counts.p_missing,
+            hashable=var_counts.hashable,
+            memory_size=var_counts.memory_size,
+            ordering=var_counts.ordering,
+            value_counts_index_sorted=var_counts.value_counts_index_sorted,
+            value_counts_without_nan=var_counts.value_counts_without_nan,
+            var_specific=init_dict,
+            is_unique=is_unique,
+            n_unique=n_unique,
+            n_distinct=n_distinct,
+            p_distinct=p_distinct,
+            p_unique=p_unique,
+            value_counts=var_counts.value_counts,
+        )
+
+
+def get_default_spark_description(
+    config: Settings, series: DataFrame, init_dict: dict
+) -> VarDescriptionSparkHashable:
+    _var_counts = VarCountsSpark(config, series)
+    return VarDescriptionSparkHashable.from_var_counts(_var_counts, init_dict)