[FEATURE] Add mean to metrics API (#10961)

Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>

Author: billdirks

great_expectations/metrics/__init__.py

@@ -1,3 +1,4 @@
 from .batch.row_count import BatchRowCount
+from .column_aggregate.mean import ColumnValuesMean
 from .column_values.non_null import ColumnValuesNonNull, ColumnValuesNonNullCount
 from .metric import Metric

great_expectations/metrics/column_aggregate/__init__.py

@@ -0,0 +1,10 @@
+from great_expectations.metrics.domain import ColumnValues
+from great_expectations.metrics.metric import Metric
+from great_expectations.metrics.metric_results import MetricResult
+
+
+class ColumnValuesMeanResult(MetricResult[float]): ...
+
+
+class ColumnValuesMean(Metric[ColumnValuesMeanResult], ColumnValues):
+    name = "column.mean"

great_expectations/metrics/column_aggregate/mean.py

@@ -0,0 +1,72 @@
+import pandas
+
+from great_expectations.metrics.column_aggregate.mean import (
+    ColumnValuesMean,
+    ColumnValuesMeanResult,
+)
+from great_expectations.metrics.metric_results import MetricErrorResult
+from tests.integration.conftest import parameterize_batch_for_data_sources
+from tests.integration.test_utils.data_source_config import (
+    BigQueryDatasourceTestConfig,
+    DatabricksDatasourceTestConfig,
+    DataSourceTestConfig,
+    MSSQLDatasourceTestConfig,
+    PandasDataFrameDatasourceTestConfig,
+    PostgreSQLDatasourceTestConfig,
+    SnowflakeDatasourceTestConfig,
+    SparkFilesystemCsvDatasourceTestConfig,
+    SqliteDatasourceTestConfig,
+)
+
+DATA_FRAME = pandas.DataFrame(
+    {
+        "id": [1, 2, 3, 4],
+        "number": [1, 2, 3, 4],
+        "string": ["a", "b", "c", "d"],
+    },
+)
+
+DATA_SOURCES_WITHOUT_SPARK_DATABRICKS_SQLITE: list[DataSourceTestConfig] = [
+    BigQueryDatasourceTestConfig(),
+    MSSQLDatasourceTestConfig(),
+    PostgreSQLDatasourceTestConfig(),
+    SnowflakeDatasourceTestConfig(),
+    PandasDataFrameDatasourceTestConfig(),
+]
+
+DATA_SOURCES: list[DataSourceTestConfig] = DATA_SOURCES_WITHOUT_SPARK_DATABRICKS_SQLITE + [
+    SparkFilesystemCsvDatasourceTestConfig(),
+    DatabricksDatasourceTestConfig(),
+    SqliteDatasourceTestConfig(),
+]
+
+
+@parameterize_batch_for_data_sources(
+    data_source_configs=DATA_SOURCES,
+    data=DATA_FRAME,
+)
+def test_mean_success(batch_for_datasource) -> None:
+    batch = batch_for_datasource
+    metric = ColumnValuesMean(batch_id=batch.id, column="number")
+    metric_result = batch.compute_metrics(metric)
+    assert isinstance(metric_result, ColumnValuesMeanResult)
+    assert metric_result.value == 2.5
+
+
+# For spark, when computing the mean, if it fails, the metric name changes from
+# `column.mean` to `column.aggregate.mean`.
+# There is a bug to track fixing this: https://greatexpectations.atlassian.net/browse/GX-448
+# For databricks, when computing the mean, any non-numeric values are ignored and the result is
+# None, which will cause a crash later when trying to set the value of the MetricResult
+# (not MetricErrorResult) to None.
+# For sqlite, when computing the mean, any non-numeric values are ignored (or maybe treated
+# as 0) so we don't an error.
+@parameterize_batch_for_data_sources(
+    data_source_configs=DATA_SOURCES_WITHOUT_SPARK_DATABRICKS_SQLITE,
+    data=DATA_FRAME,
+)
+def test_mean_failure(batch_for_datasource) -> None:
+    batch = batch_for_datasource
+    metric = ColumnValuesMean(batch_id=batch.id, column="string")
+    metric_result = batch.compute_metrics(metric)
+    assert isinstance(metric_result, MetricErrorResult)