diff --git a/qiskit_experiments/database_service/db_experiment_data.py b/qiskit_experiments/database_service/db_experiment_data.py
index 7560c507cf..1abb154dde 100644
--- a/qiskit_experiments/database_service/db_experiment_data.py
+++ b/qiskit_experiments/database_service/db_experiment_data.py
@@ -178,6 +178,17 @@ def __init__(
         self._created_in_db = False
         self._extra_data = kwargs
 
+    def _clear_results(self):
+        """Delete all currently stored analysis results and figures"""
+        # Schedule existing analysis results for deletion next save call
+        for key in self._analysis_results.keys():
+            self._deleted_analysis_results.append(key)
+        self._analysis_results = ThreadSafeOrderedDict()
+        # Schedule existing figures for deletion next save call
+        for key in self._analysis_results.keys():
+            self._deleted_figures.append(key)
+        self._figures = ThreadSafeOrderedDict()
+
     def _set_service_from_backend(self, backend: Union[Backend, BaseBackend]) -> None:
         """Set the service to be used from the input backend.
 
@@ -555,7 +566,9 @@ def delete_figure(
         return figure_key
 
     def figure(
-        self, figure_key: Union[str, int], file_name: Optional[str] = None
+        self,
+        figure_key: Union[str, int],
+        file_name: Optional[str] = None,
     ) -> Union[int, bytes]:
         """Retrieve the specified experiment figure.
 
@@ -663,7 +676,9 @@ def _retrieve_analysis_results(self, refresh: bool = False):
                 self._analysis_results[result_id] = DbAnalysisResult._from_service_data(result)
 
     def analysis_results(
-        self, index: Optional[Union[int, slice, str]] = None, refresh: bool = False
+        self,
+        index: Optional[Union[int, slice, str]] = None,
+        refresh: bool = False,
     ) -> Union[DbAnalysisResult, List[DbAnalysisResult]]:
         """Return analysis results associated with this experiment.
 
@@ -899,6 +914,12 @@ def block_for_results(self, timeout: Optional[float] = None) -> "DbExperimentDat
         Returns:
             The experiment data with finished jobs and post-processing.
         """
+        _, timeout = combined_timeout(self._wait_for_jobs, timeout)
+        _, timeout = combined_timeout(self._wait_for_callbacks, timeout)
+        return self
+
+    def _wait_for_jobs(self, timeout: Optional[float] = None):
+        """Wait for jobs to finish running"""
         # Wait for jobs to finish
         for kwargs, fut in self._job_futures.copy():
             jobs = [job.job_id() for job in kwargs["jobs"]]
@@ -914,7 +935,6 @@ def block_for_results(self, timeout: Optional[float] = None) -> "DbExperimentDat
                     "Possibly incomplete experiment data: Retrieving a job results"
                     " rased an exception."
                 )
-
         # Check job status and show warning if cancelled or error
         jobs_status = self._job_status()
         if jobs_status == "CANCELLED":
@@ -922,9 +942,13 @@ def block_for_results(self, timeout: Optional[float] = None) -> "DbExperimentDat
         elif jobs_status == "ERROR":
             LOG.warning("Possibly incomplete experiment data: A Job returned an error.")
 
+    def _wait_for_callbacks(self, timeout: Optional[float] = None):
+        """Wait for analysis callbacks to finish"""
         # Wait for analysis callbacks to finish
         if self._callback_statuses:
             for status in self._callback_statuses.values():
+                if status.status in [JobStatus.DONE, JobStatus.CANCELLED]:
+                    continue
                 LOG.info("Waiting for analysis callback %s to finish.", status.callback)
                 finished, timeout = combined_timeout(status.event.wait, timeout)
                 if not finished:
@@ -944,8 +968,6 @@ def block_for_results(self, timeout: Optional[float] = None) -> "DbExperimentDat
                 "Possibly incomplete analysis results: an analysis callback raised an error."
             )
 
-        return self
-
     def status(self) -> str:
         """Return the data processing status.
 
diff --git a/qiskit_experiments/framework/base_analysis.py b/qiskit_experiments/framework/base_analysis.py
index adfe2cf831..256009bb60 100644
--- a/qiskit_experiments/framework/base_analysis.py
+++ b/qiskit_experiments/framework/base_analysis.py
@@ -51,12 +51,16 @@ def _default_options(cls) -> Options:
     def run(
         self,
         experiment_data: ExperimentData,
+        replace_results: bool = False,
         **options,
     ) -> ExperimentData:
         """Run analysis and update ExperimentData with analysis result.
 
         Args:
             experiment_data: the experiment data to analyze.
+            replace_results: if True clear any existing analysis results and
+                             figures in the experiment data and replace with
+                             new results. See note for additional information.
             options: additional analysis options. See class documentation for
                      supported options.
 
@@ -65,6 +69,20 @@ def run(
 
         Raises:
             QiskitError: if experiment_data container is not valid for analysis.
+
+        .. note::
+            **Updating Results**
+
+            If analysis is run with ``replace_results=True`` then any analysis results
+            and figures in the experiment data will be cleared and replaced with the
+            new analysis results. Saving this experiment data will replace any
+            previously saved data in a database service using the same experiment ID.
+
+            If analysis is run with ``replace_results=False`` and the experiment data
+            being analyzed has already been saved to a database service, or already
+            contains analysis results or figures, a copy with a unique experiment ID
+            will be returned containing only the new analysis results and figures.
+            This data can then be saved as its own experiment to a database service.
         """
         if not isinstance(experiment_data, self.__experiment_data__):
             raise QiskitError(
@@ -72,6 +90,14 @@ def run(
                 f" but received {type(experiment_data).__name__}"
             )
 
+        # Make a new copy of experiment data if not updating results
+        if not replace_results and (
+            experiment_data._created_in_db
+            or experiment_data._analysis_results
+            or experiment_data._figures
+        ):
+            experiment_data = experiment_data._copy_metadata()
+
         # Get experiment device components
         if "physical_qubits" in experiment_data.metadata:
             experiment_components = [
@@ -85,21 +111,22 @@ def run(
         analysis_options.update_options(**options)
         analysis_options = analysis_options.__dict__
 
-        # Run analysis
-        results, figures = self._run_analysis(experiment_data, **analysis_options)
-
-        # Add components
-        analysis_results = [
-            self._format_analysis_result(
-                result, experiment_data.experiment_id, experiment_components
-            )
-            for result in results
-        ]
-
-        # Update experiment data with analysis results
-        experiment_data.add_analysis_results(analysis_results)
-        if figures:
-            experiment_data.add_figures(figures)
+        def run_analysis(expdata):
+            results, figures = self._run_analysis(expdata, **analysis_options)
+            # Add components
+            analysis_results = [
+                self._format_analysis_result(result, expdata.experiment_id, experiment_components)
+                for result in results
+            ]
+            # Update experiment data with analysis results
+            if replace_results:
+                experiment_data._clear_results()
+            if analysis_results:
+                expdata.add_analysis_results(analysis_results)
+            if figures:
+                expdata.add_figures(figures)
+
+        experiment_data.add_analysis_callback(run_analysis)
 
         return experiment_data
 
diff --git a/qiskit_experiments/framework/base_experiment.py b/qiskit_experiments/framework/base_experiment.py
index b771ee9d73..3d9ad1ac60 100644
--- a/qiskit_experiments/framework/base_experiment.py
+++ b/qiskit_experiments/framework/base_experiment.py
@@ -317,7 +317,7 @@ def run(
 
         # Optionally run analysis
         if analysis and self.__analysis_class__ is not None:
-            experiment_data.add_analysis_callback(self.run_analysis)
+            self.run_analysis(experiment_data)
 
         # Return the ExperimentData future
         return experiment_data
@@ -326,11 +326,18 @@ def _initialize_experiment_data(self) -> ExperimentData:
         """Initialize the return data container for the experiment run"""
         return self.__experiment_data__(experiment=self)
 
-    def run_analysis(self, experiment_data: ExperimentData, **options) -> ExperimentData:
+    def run_analysis(
+        self, experiment_data: ExperimentData, replace_results: bool = False, **options
+    ) -> ExperimentData:
         """Run analysis and update ExperimentData with analysis result.
 
+        See :meth:`BaseAnalysis.run` for additional information.
+
         Args:
             experiment_data: the experiment data to analyze.
+            replace_results: if True clear any existing analysis results and
+                             figures in the experiment data and replace with
+                             new results.
             options: additional analysis options. Any values set here will
                      override the value from :meth:`analysis_options`
                      for the current run.
@@ -348,7 +355,7 @@ def run_analysis(self, experiment_data: ExperimentData, **options) -> Experiment
 
         # Run analysis
         analysis = self.analysis()
-        analysis.run(experiment_data, **analysis_options)
+        analysis.run(experiment_data, replace_results=replace_results, **analysis_options)
         return experiment_data
 
     def _run_jobs(self, circuits: List[QuantumCircuit], **run_options) -> List[BaseJob]:
diff --git a/qiskit_experiments/framework/composite/composite_experiment_data.py b/qiskit_experiments/framework/composite/composite_experiment_data.py
index 1b0eefe06f..846107a587 100644
--- a/qiskit_experiments/framework/composite/composite_experiment_data.py
+++ b/qiskit_experiments/framework/composite/composite_experiment_data.py
@@ -18,6 +18,7 @@
 from qiskit.exceptions import QiskitError
 from qiskit_experiments.framework.experiment_data import ExperimentData
 from qiskit_experiments.database_service import DatabaseServiceV1
+from qiskit_experiments.database_service.utils import combined_timeout
 
 
 class CompositeExperimentData(ExperimentData):
@@ -189,3 +190,8 @@ def _copy_metadata(
             comp.experiment_id for comp in new_instance.component_experiment_data()
         ]
         return new_instance
+
+    def block_for_results(self, timeout: Optional[float] = None):
+        _, timeout = combined_timeout(super().block_for_results, timeout)
+        for subdata in self._components:
+            _, timeout = combined_timeout(subdata.block_for_results, timeout)
diff --git a/releasenotes/notes/analysis-run-b4ba83436a562a01.yaml b/releasenotes/notes/analysis-run-b4ba83436a562a01.yaml
new file mode 100644
index 0000000000..a00b935639
--- /dev/null
+++ b/releasenotes/notes/analysis-run-b4ba83436a562a01.yaml
@@ -0,0 +1,24 @@
+---
+features:
+  - |
+    Added the ``replace_results`` kwarg to
+    :meth:`~qiskit_experiments.framework.BaseAnalysis.run` with default
+    value of ``replace_results=False``.
+    
+    If analysis is run with ``replace_results=True`` then any analysis results
+    and figures in the experiment data will be cleared and replaced with the
+    new analysis results. Saving this experiment data will replace any
+    previously saved data in a database service using the same experiment ID.
+
+    If analysis is run with ``replace_results=False`` and the experiment data
+    being analyzed has already been saved to a database service, or already
+    contains analysis results or figures, a copy with a unique experiment ID
+    will be returned containing only the new analysis results and figures.
+    This data can then be saved as its own experiment to a database service.
+upgrade:
+  - |
+    Changed :meth:`~qiskit_experiments.framework.BaseAnalysis.run` to run
+    asynchronously using the 
+    :meth:`~qiskit_experiments.framework.ExperimentData.add_analysis_callback`.
+    Previously analysis was only run asynchronously if it was done as part of
+    an experiments :meth:`~qiskit_experiments.framework.BaseExperiment.run`.
diff --git a/test/calibration/experiments/test_rabi.py b/test/calibration/experiments/test_rabi.py
index c75a1203b7..53a6e166b4 100644
--- a/test/calibration/experiments/test_rabi.py
+++ b/test/calibration/experiments/test_rabi.py
@@ -264,8 +264,10 @@ def test_good_analysis(self):
 
         data_processor = DataProcessor("counts", [Probability(outcome="1")])
 
-        experiment_data = OscillationAnalysis().run(
-            experiment_data, data_processor=data_processor, plot=False
+        experiment_data = (
+            OscillationAnalysis()
+            .run(experiment_data, data_processor=data_processor, plot=False)
+            .block_for_results()
         )
         result = experiment_data.analysis_results()
         self.assertEqual(result[0].quality, "good")
@@ -282,8 +284,10 @@ def test_bad_analysis(self):
 
         data_processor = DataProcessor("counts", [Probability(outcome="1")])
 
-        experiment_data = OscillationAnalysis().run(
-            experiment_data, data_processor=data_processor, plot=False
+        experiment_data = (
+            OscillationAnalysis()
+            .run(experiment_data, data_processor=data_processor, plot=False)
+            .block_for_results()
         )
         result = experiment_data.analysis_results()
 
diff --git a/test/fake_experiment.py b/test/fake_experiment.py
index f5abddba5a..b4eb6a2c0c 100644
--- a/test/fake_experiment.py
+++ b/test/fake_experiment.py
@@ -12,7 +12,8 @@
 
 """A FakeExperiment for testing."""
 
-from qiskit_experiments.framework import BaseExperiment, BaseAnalysis, Options
+import numpy as np
+from qiskit_experiments.framework import BaseExperiment, BaseAnalysis, Options, AnalysisResultData
 
 
 class FakeAnalysis(BaseAnalysis):
@@ -21,7 +22,12 @@ class FakeAnalysis(BaseAnalysis):
     """
 
     def _run_analysis(self, experiment_data, **options):
-        return [], None
+        seed = options.get("seed", None)
+        rng = np.random.default_rng(seed=seed)
+        analysis_results = [
+            AnalysisResultData(f"result_{i}", value) for i, value in enumerate(rng.random(3))
+        ]
+        return analysis_results, None
 
 
 class FakeExperiment(BaseExperiment):
diff --git a/test/quantum_volume/test_qv.py b/test/quantum_volume/test_qv.py
index 1f615c36d8..c095ef4cd9 100644
--- a/test/quantum_volume/test_qv.py
+++ b/test/quantum_volume/test_qv.py
@@ -109,7 +109,7 @@ def test_qv_sigma_decreasing(self):
         result_data1 = expdata1.analysis_results(0)
         expdata2 = qv_exp.run(backend, analysis=False).block_for_results()
         expdata2.add_data(expdata1.data())
-        qv_exp.run_analysis(expdata2)
+        qv_exp.run_analysis(expdata2).block_for_results()
         result_data2 = expdata2.analysis_results(0)
 
         self.assertTrue(result_data1.extra["trials"] == 2, "number of trials is incorrect")
@@ -139,7 +139,7 @@ def test_qv_failure_insufficient_trials(self):
         exp_data = ExperimentData(experiment=qv_exp, backend=backend)
         exp_data.add_data(insufficient_trials_data)
 
-        qv_exp.run_analysis(exp_data)
+        qv_exp.run_analysis(exp_data).block_for_results()
         qv_result = exp_data.analysis_results(1)
         self.assertTrue(
             qv_result.extra["success"] is False and qv_result.value == 1,
@@ -163,7 +163,7 @@ def test_qv_failure_insufficient_hop(self):
         exp_data = ExperimentData(experiment=qv_exp, backend=backend)
         exp_data.add_data(insufficient_hop_data)
 
-        qv_exp.run_analysis(exp_data)
+        qv_exp.run_analysis(exp_data).block_for_results()
         qv_result = exp_data.analysis_results(1)
         self.assertTrue(
             qv_result.extra["success"] is False and qv_result.value == 1,
@@ -188,7 +188,7 @@ def test_qv_failure_insufficient_confidence(self):
         exp_data = ExperimentData(experiment=qv_exp, backend=backend)
         exp_data.add_data(insufficient_confidence_data)
 
-        qv_exp.run_analysis(exp_data)
+        qv_exp.run_analysis(exp_data).block_for_results()
         qv_result = exp_data.analysis_results(1)
         self.assertTrue(
             qv_result.extra["success"] is False and qv_result.value == 1,
@@ -212,7 +212,7 @@ def test_qv_success(self):
         exp_data = ExperimentData(experiment=qv_exp, backend=backend)
         exp_data.add_data(successful_data)
 
-        qv_exp.run_analysis(exp_data)
+        qv_exp.run_analysis(exp_data).block_for_results()
         results_json_file = "qv_result_moderate_noise_300_trials.json"
         with open(os.path.join(dir_name, results_json_file), "r") as json_file:
             successful_results = json.load(json_file, cls=ExperimentDecoder)
diff --git a/test/randomized_benchmarking/test_rb_analysis.py b/test/randomized_benchmarking/test_rb_analysis.py
index e768b34e33..a1806dbce5 100644
--- a/test/randomized_benchmarking/test_rb_analysis.py
+++ b/test/randomized_benchmarking/test_rb_analysis.py
@@ -205,7 +205,7 @@ def _load_rb_data(self, rb_exp_data_file_name: str):
             ((0, 1), "cx"): 1,
         }
         rb_exp.set_analysis_options(gate_error_ratio=gate_error_ratio)
-        analysis_results = rb_exp.run_analysis(expdata1)
+        analysis_results = rb_exp.run_analysis(expdata1).block_for_results()
         return data, analysis_results
 
 
@@ -260,7 +260,7 @@ def _load_rb_data(self, rb_exp_data_file_name: str):
             ((0, 1), "cx"): 1,
         }
         rb_exp.set_analysis_options(gate_error_ratio=gate_error_ratio)
-        analysis_results = rb_exp.run_analysis(expdata1)
+        analysis_results = rb_exp.run_analysis(expdata1).block_for_results()
         return data, analysis_results
 
     def test_interleaved_rb_analysis_test(self):
diff --git a/test/test_framework.py b/test/test_framework.py
index 389b3a4eb6..851fa3a1d0 100644
--- a/test/test_framework.py
+++ b/test/test_framework.py
@@ -13,12 +13,12 @@
 """Tests for base experiment framework."""
 
 from test.fake_backend import FakeBackend
-from test.fake_experiment import FakeExperiment
-
+from test.fake_experiment import FakeExperiment, FakeAnalysis
 import ddt
 
 from qiskit import QuantumCircuit
 from qiskit.test import QiskitTestCase
+from qiskit_experiments.framework import ExperimentData
 
 
 @ddt.ddt
@@ -53,3 +53,24 @@ def circuits(self):
             if num_circuits % max_experiments:
                 num_jobs += 1
         self.assertEqual(len(job_ids), num_jobs)
+
+    def test_analysis_replace_results_true(self):
+        """Test running analysis with replace_results=True"""
+        analysis = FakeAnalysis()
+        expdata1 = analysis.run(ExperimentData(), seed=54321).block_for_results()
+        result_ids = [res.result_id for res in expdata1.analysis_results()]
+        expdata2 = analysis.run(expdata1, replace_results=True, seed=12345).block_for_results()
+
+        self.assertEqual(expdata1, expdata2)
+        self.assertEqual(expdata1.analysis_results(), expdata2.analysis_results())
+        self.assertEqual(result_ids, list(expdata2._deleted_analysis_results))
+
+    def test_analysis_replace_results_false(self):
+        """Test running analysis with replace_results=False"""
+        analysis = FakeAnalysis()
+        expdata1 = analysis.run(ExperimentData(), seed=54321).block_for_results()
+        expdata2 = analysis.run(expdata1, replace_results=False, seed=12345).block_for_results()
+
+        self.assertNotEqual(expdata1, expdata2)
+        self.assertNotEqual(expdata1.experiment_id, expdata2.experiment_id)
+        self.assertNotEqual(expdata1.analysis_results(), expdata2.analysis_results())
diff --git a/test/test_t2ramsey.py b/test/test_t2ramsey.py
index 4e8f7357fc..f4596c718f 100644
--- a/test/test_t2ramsey.py
+++ b/test/test_t2ramsey.py
@@ -179,7 +179,7 @@ def test_t2ramsey_concat_2_experiments(self):
         exp1.set_analysis_options(user_p0=default_p0)
         expdata1 = exp1.run(backend=backend, analysis=False, shots=1000).block_for_results()
         expdata1.add_data(expdata0.data())
-        exp1.run_analysis(expdata1)
+        exp1.run_analysis(expdata1).block_for_results()
         results1 = expdata1.analysis_results()
 
         self.assertAlmostEqual(