[FEATURE] Add run_profiler_on_data method to DataContext

great_expectations/data_context/data_context.py

@@ -3348,6 +3348,23 @@ def run_profiler_with_dynamic_arguments(
             include_citation=include_citation,
         )
 
+    @usage_statistics_enabled_method(
+        event_name="data_context.run_profiler_on_data",
+    )
+    def run_profiler_on_data(
+        self,
+        name: Optional[str] = None,
+        ge_cloud_id: Optional[str] = None,
+        batch_request: Optional[Union[dict, BatchRequest, RuntimeBatchRequest]] = None,
+    ) -> ExpectationSuite:
+        return RuleBasedProfiler.run_profiler_on_data(
+            data_context=self,
+            profiler_store=self.profiler_store,
+            batch_request=batch_request,
+            name=name,
+            ge_cloud_id=ge_cloud_id,
+        )
+
     def test_yaml_config(
         self,
         yaml_config: str,

great_expectations/rule_based_profiler/rule_based_profiler.py

@@ -7,6 +7,7 @@
     BatchRequest,
     RuntimeBatchRequest,
     batch_request_contains_batch_data,
+    get_batch_request_from_acceptable_arguments,
 )
 from great_expectations.core.config_peer import ConfigPeer
 from great_expectations.core.expectation_configuration import ExpectationConfiguration
@@ -25,6 +26,9 @@
     expectationConfigurationBuilderConfigSchema,
     parameterBuilderConfigSchema,
 )
+from great_expectations.rule_based_profiler.domain_builder.column_domain_builder import (
+    ColumnDomainBuilder,
+)
 from great_expectations.rule_based_profiler.domain_builder.domain_builder import (
     DomainBuilder,
 )
@@ -309,6 +313,39 @@ def run(
 
         return expectation_suite
 
+    def reconcile_batch_requests_in_builders(
+        self, batch_request: Union[dict, BatchRequest, RuntimeBatchRequest]
+    ) -> None:
+        """
+        Profiler "batch_request" reconciliation involves combining existing Profiler state, instantiated from Profiler configuration
+        (e.g., stored in a YAML file managed by the Profiler store), with the batch request overrides, provided at run time.
+
+        The provided batch request is propagated to the following relevant Builders attributes (as applicable):
+            - ParameterBuilders
+            - ColumnDomainBuilder
+              - We default to the latest value as a sensible default
+
+        The reconciliation logic for "batch_request" is of the "replace" nature: the provided data is consistently applied, regardless
+        of existing Builder state.
+
+        Args:
+            batch_request: Data provided at runtime used to hydrate nested builder attributes
+        """
+        if isinstance(batch_request, dict):
+            batch_request = get_batch_request_from_acceptable_arguments(**batch_request)
+
+        # TODO(cdkini): Let's use setters
+        for rule in self.rules:
+            domain_builder = rule.domain_builder
+            if isinstance(domain_builder, ColumnDomainBuilder):
+                domain_builder._batch_request = copy.deepcopy(batch_request)
+                domain_builder._batch_request.data_connector_query = {"index": -1}
+
+            parameter_builders = rule.parameter_builders
+            if parameter_builders:
+                for parameter_builder in parameter_builders:
+                    parameter_builder._batch_request = copy.deepcopy(batch_request)
+
     def reconcile_profiler_variables(
         self, variables: Optional[Dict[str, Any]] = None
     ) -> Optional[ParameterContainer]:
@@ -816,6 +853,25 @@ def run_profiler(
 
         return result
 
+    @staticmethod
+    def run_profiler_on_data(
+        data_context: "DataContext",  # noqa: F821
+        profiler_store: ProfilerStore,
+        batch_request: Union[dict, BatchRequest, RuntimeBatchRequest],
+        name: Optional[str] = None,
+        ge_cloud_id: Optional[str] = None,
+    ) -> ExpectationSuite:
+        profiler: RuleBasedProfiler = RuleBasedProfiler.get_profiler(
+            data_context=data_context,
+            profiler_store=profiler_store,
+            name=name,
+            ge_cloud_id=ge_cloud_id,
+        )
+        profiler.reconcile_batch_requests_in_builders(batch_request)
+
+        result: ExpectationSuite = profiler.run()
+        return result
+
     @staticmethod
     def add_profiler(
         config: RuleBasedProfilerConfig,

tests/data_context/test_data_context_profilers.py

@@ -79,3 +79,7 @@ def test_run_profiler_with_dynamic_arguments_emits_proper_usage_stats(
             }
         )
     ]
+
+
+def test_run_profiler_on_data_emits_proper_usage_stats():
+    pass  # TBD

tests/rule_based_profiler/test_rule_based_profiler.py

@@ -701,3 +701,15 @@ def test_list_profilers_in_cloud_mode(mock_profiler_store: mock.MagicMock):
 
     assert res == keys
     assert store.list_keys.called
+
+
+def test_reconcile_batch_requests_in_builders_replaces_batch_requests():
+    pass  # TBD
+
+
+def test_reconcile_batch_requests_in_builders_does_not_replace_batch_requests():
+    pass  # TBD
+
+
+def test_run_profiler_on_data():
+    pass  # TBD