diff --git a/docs/development/plugins/expressions/public/kibana-plugin-plugins-expressions-public.expressionfunctiondefinitions.cumulative_sum.md b/docs/development/plugins/expressions/public/kibana-plugin-plugins-expressions-public.expressionfunctiondefinitions.cumulative_sum.md
new file mode 100644
index 0000000000000..ad1de0cc5f45b
--- /dev/null
+++ b/docs/development/plugins/expressions/public/kibana-plugin-plugins-expressions-public.expressionfunctiondefinitions.cumulative_sum.md
@@ -0,0 +1,11 @@
+<!-- Do not edit this file. It is automatically generated by API Documenter. -->
+
+[Home](./index.md) &gt; [kibana-plugin-plugins-expressions-public](./kibana-plugin-plugins-expressions-public.md) &gt; [ExpressionFunctionDefinitions](./kibana-plugin-plugins-expressions-public.expressionfunctiondefinitions.md) &gt; [cumulative\_sum](./kibana-plugin-plugins-expressions-public.expressionfunctiondefinitions.cumulative_sum.md)
+
+## ExpressionFunctionDefinitions.cumulative\_sum property
+
+<b>Signature:</b>
+
+```typescript
+cumulative_sum: ExpressionFunctionCumulativeSum;
+```
diff --git a/docs/development/plugins/expressions/public/kibana-plugin-plugins-expressions-public.expressionfunctiondefinitions.md b/docs/development/plugins/expressions/public/kibana-plugin-plugins-expressions-public.expressionfunctiondefinitions.md
index 914c5d6ebe2f6..d1703a1e019e6 100644
--- a/docs/development/plugins/expressions/public/kibana-plugin-plugins-expressions-public.expressionfunctiondefinitions.md
+++ b/docs/development/plugins/expressions/public/kibana-plugin-plugins-expressions-public.expressionfunctiondefinitions.md
@@ -17,6 +17,7 @@ export interface ExpressionFunctionDefinitions
 |  Property | Type | Description |
 |  --- | --- | --- |
 |  [clog](./kibana-plugin-plugins-expressions-public.expressionfunctiondefinitions.clog.md) | <code>ExpressionFunctionClog</code> |  |
+|  [cumulative\_sum](./kibana-plugin-plugins-expressions-public.expressionfunctiondefinitions.cumulative_sum.md) | <code>ExpressionFunctionCumulativeSum</code> |  |
 |  [font](./kibana-plugin-plugins-expressions-public.expressionfunctiondefinitions.font.md) | <code>ExpressionFunctionFont</code> |  |
 |  [kibana\_context](./kibana-plugin-plugins-expressions-public.expressionfunctiondefinitions.kibana_context.md) | <code>ExpressionFunctionKibanaContext</code> |  |
 |  [kibana](./kibana-plugin-plugins-expressions-public.expressionfunctiondefinitions.kibana.md) | <code>ExpressionFunctionKibana</code> |  |
diff --git a/docs/development/plugins/expressions/server/kibana-plugin-plugins-expressions-server.expressionfunctiondefinitions.cumulative_sum.md b/docs/development/plugins/expressions/server/kibana-plugin-plugins-expressions-server.expressionfunctiondefinitions.cumulative_sum.md
new file mode 100644
index 0000000000000..2fb8cde92e877
--- /dev/null
+++ b/docs/development/plugins/expressions/server/kibana-plugin-plugins-expressions-server.expressionfunctiondefinitions.cumulative_sum.md
@@ -0,0 +1,11 @@
+<!-- Do not edit this file. It is automatically generated by API Documenter. -->
+
+[Home](./index.md) &gt; [kibana-plugin-plugins-expressions-server](./kibana-plugin-plugins-expressions-server.md) &gt; [ExpressionFunctionDefinitions](./kibana-plugin-plugins-expressions-server.expressionfunctiondefinitions.md) &gt; [cumulative\_sum](./kibana-plugin-plugins-expressions-server.expressionfunctiondefinitions.cumulative_sum.md)
+
+## ExpressionFunctionDefinitions.cumulative\_sum property
+
+<b>Signature:</b>
+
+```typescript
+cumulative_sum: ExpressionFunctionCumulativeSum;
+```
diff --git a/docs/development/plugins/expressions/server/kibana-plugin-plugins-expressions-server.expressionfunctiondefinitions.md b/docs/development/plugins/expressions/server/kibana-plugin-plugins-expressions-server.expressionfunctiondefinitions.md
index 71cd0b98a68c2..05b4ddce4ccde 100644
--- a/docs/development/plugins/expressions/server/kibana-plugin-plugins-expressions-server.expressionfunctiondefinitions.md
+++ b/docs/development/plugins/expressions/server/kibana-plugin-plugins-expressions-server.expressionfunctiondefinitions.md
@@ -17,6 +17,7 @@ export interface ExpressionFunctionDefinitions
 |  Property | Type | Description |
 |  --- | --- | --- |
 |  [clog](./kibana-plugin-plugins-expressions-server.expressionfunctiondefinitions.clog.md) | <code>ExpressionFunctionClog</code> |  |
+|  [cumulative\_sum](./kibana-plugin-plugins-expressions-server.expressionfunctiondefinitions.cumulative_sum.md) | <code>ExpressionFunctionCumulativeSum</code> |  |
 |  [font](./kibana-plugin-plugins-expressions-server.expressionfunctiondefinitions.font.md) | <code>ExpressionFunctionFont</code> |  |
 |  [kibana\_context](./kibana-plugin-plugins-expressions-server.expressionfunctiondefinitions.kibana_context.md) | <code>ExpressionFunctionKibanaContext</code> |  |
 |  [kibana](./kibana-plugin-plugins-expressions-server.expressionfunctiondefinitions.kibana.md) | <code>ExpressionFunctionKibana</code> |  |
diff --git a/src/plugins/expressions/common/expression_functions/specs/cumulative_sum.ts b/src/plugins/expressions/common/expression_functions/specs/cumulative_sum.ts
new file mode 100644
index 0000000000000..970015638794f
--- /dev/null
+++ b/src/plugins/expressions/common/expression_functions/specs/cumulative_sum.ts
@@ -0,0 +1,170 @@
+/*
+ * Licensed to Elasticsearch B.V. under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch B.V. licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import { i18n } from '@kbn/i18n';
+import { ExpressionFunctionDefinition } from '../types';
+import { Datatable, DatatableRow } from '../../expression_types';
+
+export interface CumulativeSumArgs {
+  by?: string[];
+  inputColumnId: string;
+  outputColumnId: string;
+  outputColumnName?: string;
+}
+
+export type ExpressionFunctionCumulativeSum = ExpressionFunctionDefinition<
+  'cumulative_sum',
+  Datatable,
+  CumulativeSumArgs,
+  Datatable
+>;
+
+/**
+ * Returns a string identifying the group of a row by a list of columns to group by
+ */
+function getBucketIdentifier(row: DatatableRow, groupColumns?: string[]) {
+  return (groupColumns || [])
+    .map((groupColumnId) => (row[groupColumnId] == null ? '' : String(row[groupColumnId])))
+    .join('|');
+}
+
+/**
+ * Calculates the cumulative sum of a specified column in the data table.
+ *
+ * Also supports multiple series in a single data table - use the `by` argument
+ * to specify the columns to split the calculation by.
+ * For each unique combination of all `by` columns a separate cumulative sum will be calculated.
+ * The order of rows won't be changed - this function is not modifying any existing columns, it's only
+ * adding the specified `outputColumnId` column to every row of the table without adding or removing rows.
+ *
+ * Behavior:
+ * * Will write the cumulative sum of `inputColumnId` into `outputColumnId`
+ * * If provided will use `outputColumnName` as name for the newly created column. Otherwise falls back to `outputColumnId`
+ * * Cumulative sums always start with 0, a cell will contain its own value plus the values of
+ *   all cells of the same series further up in the table.
+ *
+ * Edge cases:
+ * * Will return the input table if `inputColumnId` does not exist
+ * * Will throw an error if `outputColumnId` exists already in provided data table
+ * * If the row value contains `null` or `undefined`, it will be ignored and overwritten with the cumulative sum of
+ *   all cells of the same series further up in the table.
+ * * For all values besides `null` and `undefined`, the value will be cast to a number before it's added to the
+ *   cumulative sum of the current series - if this results in `NaN` (like in case of objects), all cells of the
+ *   current series will be set to `NaN`.
+ * * To determine separate series defined by the `by` columns, the values of these columns will be cast to strings
+ *   before comparison. If the values are objects, the return value of their `toString` method will be used for comparison.
+ *   Missing values (`null` and `undefined`) will be treated as empty strings.
+ */
+export const cumulativeSum: ExpressionFunctionCumulativeSum = {
+  name: 'cumulative_sum',
+  type: 'datatable',
+
+  inputTypes: ['datatable'],
+
+  help: i18n.translate('expressions.functions.cumulativeSum.help', {
+    defaultMessage: 'Calculates the cumulative sum of a column in a data table',
+  }),
+
+  args: {
+    by: {
+      help: i18n.translate('expressions.functions.cumulativeSum.args.byHelpText', {
+        defaultMessage: 'Column to split the cumulative sum calculation by',
+      }),
+      multi: true,
+      types: ['string'],
+      required: false,
+    },
+    inputColumnId: {
+      help: i18n.translate('expressions.functions.cumulativeSum.args.inputColumnIdHelpText', {
+        defaultMessage: 'Column to calculate the cumulative sum of',
+      }),
+      types: ['string'],
+      required: true,
+    },
+    outputColumnId: {
+      help: i18n.translate('expressions.functions.cumulativeSum.args.outputColumnIdHelpText', {
+        defaultMessage: 'Column to store the resulting cumulative sum in',
+      }),
+      types: ['string'],
+      required: true,
+    },
+    outputColumnName: {
+      help: i18n.translate('expressions.functions.cumulativeSum.args.outputColumnNameHelpText', {
+        defaultMessage: 'Name of the column to store the resulting cumulative sum in',
+      }),
+      types: ['string'],
+      required: false,
+    },
+  },
+
+  fn(input, { by, inputColumnId, outputColumnId, outputColumnName }) {
+    if (input.columns.some((column) => column.id === outputColumnId)) {
+      throw new Error(
+        i18n.translate('expressions.functions.cumulativeSum.columnConflictMessage', {
+          defaultMessage:
+            'Specified outputColumnId {columnId} already exists. Please pick another column id.',
+          values: {
+            columnId: outputColumnId,
+          },
+        })
+      );
+    }
+
+    const inputColumnDefinition = input.columns.find((column) => column.id === inputColumnId);
+
+    if (!inputColumnDefinition) {
+      return input;
+    }
+
+    const outputColumnDefinition = {
+      ...inputColumnDefinition,
+      id: outputColumnId,
+      name: outputColumnName || outputColumnId,
+    };
+
+    const resultColumns = [...input.columns];
+    // add output column after input column in the table
+    resultColumns.splice(
+      resultColumns.indexOf(inputColumnDefinition) + 1,
+      0,
+      outputColumnDefinition
+    );
+
+    const accumulators: Partial<Record<string, number>> = {};
+    return {
+      ...input,
+      columns: resultColumns,
+      rows: input.rows.map((row) => {
+        const newRow = { ...row };
+
+        const bucketIdentifier = getBucketIdentifier(row, by);
+        const accumulatorValue = accumulators[bucketIdentifier] ?? 0;
+        const currentValue = newRow[inputColumnId];
+        if (currentValue != null) {
+          newRow[outputColumnId] = Number(currentValue) + accumulatorValue;
+          accumulators[bucketIdentifier] = newRow[outputColumnId];
+        } else {
+          newRow[outputColumnId] = accumulatorValue;
+        }
+
+        return newRow;
+      }),
+    };
+  },
+};
diff --git a/src/plugins/expressions/common/expression_functions/specs/index.ts b/src/plugins/expressions/common/expression_functions/specs/index.ts
index 5b9562dae5f2e..aadea5882b9c0 100644
--- a/src/plugins/expressions/common/expression_functions/specs/index.ts
+++ b/src/plugins/expressions/common/expression_functions/specs/index.ts
@@ -25,6 +25,7 @@ import { variableSet } from './var_set';
 import { variable } from './var';
 import { AnyExpressionFunctionDefinition } from '../types';
 import { theme } from './theme';
+import { cumulativeSum } from './cumulative_sum';
 
 export const functionSpecs: AnyExpressionFunctionDefinition[] = [
   clog,
@@ -34,6 +35,7 @@ export const functionSpecs: AnyExpressionFunctionDefinition[] = [
   variableSet,
   variable,
   theme,
+  cumulativeSum,
 ];
 
 export * from './clog';
@@ -43,3 +45,4 @@ export * from './kibana_context';
 export * from './var_set';
 export * from './var';
 export * from './theme';
+export * from './cumulative_sum';
diff --git a/src/plugins/expressions/common/expression_functions/specs/tests/cumulative_sum.test.ts b/src/plugins/expressions/common/expression_functions/specs/tests/cumulative_sum.test.ts
new file mode 100644
index 0000000000000..037b3ddc25f89
--- /dev/null
+++ b/src/plugins/expressions/common/expression_functions/specs/tests/cumulative_sum.test.ts
@@ -0,0 +1,361 @@
+/*
+ * Licensed to Elasticsearch B.V. under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch B.V. licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import { functionWrapper } from './utils';
+import { cumulativeSum, CumulativeSumArgs } from '../cumulative_sum';
+import { ExecutionContext } from '../../../execution/types';
+import { Datatable } from '../../../expression_types/specs/datatable';
+
+describe('interpreter/functions#cumulative_sum', () => {
+  const fn = functionWrapper(cumulativeSum);
+  const runFn = (input: Datatable, args: CumulativeSumArgs) =>
+    fn(input, args, {} as ExecutionContext) as Datatable;
+
+  it('calculates cumulative sum', () => {
+    const result = runFn(
+      {
+        type: 'datatable',
+        columns: [{ id: 'val', name: 'val', meta: { type: 'number' } }],
+        rows: [{ val: 5 }, { val: 7 }, { val: 3 }, { val: 2 }],
+      },
+      { inputColumnId: 'val', outputColumnId: 'output' }
+    );
+    expect(result.columns).toContainEqual({
+      id: 'output',
+      name: 'output',
+      meta: { type: 'number' },
+    });
+    expect(result.rows.map((row) => row.output)).toEqual([5, 12, 15, 17]);
+  });
+
+  it('replaces null or undefined data with zeroes until there is real data', () => {
+    const result = runFn(
+      {
+        type: 'datatable',
+        columns: [{ id: 'val', name: 'val', meta: { type: 'number' } }],
+        rows: [{}, { val: null }, { val: undefined }, { val: 1 }],
+      },
+      { inputColumnId: 'val', outputColumnId: 'output' }
+    );
+    expect(result.columns).toContainEqual({
+      id: 'output',
+      name: 'output',
+      meta: { type: 'number' },
+    });
+    expect(result.rows.map((row) => row.output)).toEqual([0, 0, 0, 1]);
+  });
+
+  it('calculates cumulative sum for multiple series', () => {
+    const result = runFn(
+      {
+        type: 'datatable',
+        columns: [
+          { id: 'val', name: 'val', meta: { type: 'number' } },
+          { id: 'split', name: 'split', meta: { type: 'string' } },
+        ],
+        rows: [
+          { val: 1, split: 'A' },
+          { val: 2, split: 'B' },
+          { val: 3, split: 'B' },
+          { val: 4, split: 'A' },
+          { val: 5, split: 'A' },
+          { val: 6, split: 'A' },
+          { val: 7, split: 'B' },
+          { val: 8, split: 'B' },
+        ],
+      },
+      { inputColumnId: 'val', outputColumnId: 'output', by: ['split'] }
+    );
+
+    expect(result.rows.map((row) => row.output)).toEqual([
+      1,
+      2,
+      2 + 3,
+      1 + 4,
+      1 + 4 + 5,
+      1 + 4 + 5 + 6,
+      2 + 3 + 7,
+      2 + 3 + 7 + 8,
+    ]);
+  });
+
+  it('treats missing split column as separate series', () => {
+    const result = runFn(
+      {
+        type: 'datatable',
+        columns: [
+          { id: 'val', name: 'val', meta: { type: 'number' } },
+          { id: 'split', name: 'split', meta: { type: 'string' } },
+        ],
+        rows: [
+          { val: 1, split: 'A' },
+          { val: 2, split: 'B' },
+          { val: 3 },
+          { val: 4, split: 'A' },
+          { val: 5 },
+          { val: 6, split: 'A' },
+          { val: 7, split: 'B' },
+          { val: 8, split: 'B' },
+        ],
+      },
+      { inputColumnId: 'val', outputColumnId: 'output', by: ['split'] }
+    );
+    expect(result.rows.map((row) => row.output)).toEqual([
+      1,
+      2,
+      3,
+      1 + 4,
+      3 + 5,
+      1 + 4 + 6,
+      2 + 7,
+      2 + 7 + 8,
+    ]);
+  });
+
+  it('treats null like undefined and empty string for split columns', () => {
+    const result = runFn(
+      {
+        type: 'datatable',
+        columns: [
+          { id: 'val', name: 'val', meta: { type: 'number' } },
+          { id: 'split', name: 'split', meta: { type: 'string' } },
+        ],
+        rows: [
+          { val: 1, split: 'A' },
+          { val: 2, split: 'B' },
+          { val: 3 },
+          { val: 4, split: 'A' },
+          { val: 5 },
+          { val: 6, split: 'A' },
+          { val: 7, split: null },
+          { val: 8, split: 'B' },
+          { val: 9, split: '' },
+        ],
+      },
+      { inputColumnId: 'val', outputColumnId: 'output', by: ['split'] }
+    );
+    expect(result.rows.map((row) => row.output)).toEqual([
+      1,
+      2,
+      3,
+      1 + 4,
+      3 + 5,
+      1 + 4 + 6,
+      3 + 5 + 7,
+      2 + 8,
+      3 + 5 + 7 + 9,
+    ]);
+  });
+
+  it('calculates cumulative sum for multiple series by multiple split columns', () => {
+    const result = runFn(
+      {
+        type: 'datatable',
+        columns: [
+          { id: 'val', name: 'val', meta: { type: 'number' } },
+          { id: 'split', name: 'split', meta: { type: 'string' } },
+          { id: 'split2', name: 'split2', meta: { type: 'string' } },
+        ],
+        rows: [
+          { val: 1, split: 'A', split2: 'C' },
+          { val: 2, split: 'B', split2: 'C' },
+          { val: 3, split2: 'C' },
+          { val: 4, split: 'A', split2: 'C' },
+          { val: 5 },
+          { val: 6, split: 'A', split2: 'D' },
+          { val: 7, split: 'B', split2: 'D' },
+          { val: 8, split: 'B', split2: 'D' },
+        ],
+      },
+      { inputColumnId: 'val', outputColumnId: 'output', by: ['split', 'split2'] }
+    );
+    expect(result.rows.map((row) => row.output)).toEqual([1, 2, 3, 1 + 4, 5, 6, 7, 7 + 8]);
+  });
+
+  it('splits separate series by the string representation of the cell values', () => {
+    const result = runFn(
+      {
+        type: 'datatable',
+        columns: [
+          { id: 'val', name: 'val', meta: { type: 'number' } },
+          { id: 'split', name: 'split', meta: { type: 'string' } },
+        ],
+        rows: [
+          { val: 1, split: { anObj: 3 } },
+          { val: 2, split: { anotherObj: 5 } },
+          { val: 10, split: 5 },
+          { val: 11, split: '5' },
+        ],
+      },
+      { inputColumnId: 'val', outputColumnId: 'output', by: ['split'] }
+    );
+
+    expect(result.rows.map((row) => row.output)).toEqual([1, 1 + 2, 10, 21]);
+  });
+
+  it('casts values to number before calculating cumulative sum', () => {
+    const result = runFn(
+      {
+        type: 'datatable',
+        columns: [{ id: 'val', name: 'val', meta: { type: 'number' } }],
+        rows: [{ val: 5 }, { val: '7' }, { val: '3' }, { val: 2 }],
+      },
+      { inputColumnId: 'val', outputColumnId: 'output' }
+    );
+    expect(result.rows.map((row) => row.output)).toEqual([5, 12, 15, 17]);
+  });
+
+  it('casts values to number before calculating cumulative sum for NaN like values', () => {
+    const result = runFn(
+      {
+        type: 'datatable',
+        columns: [{ id: 'val', name: 'val', meta: { type: 'number' } }],
+        rows: [{ val: 5 }, { val: '7' }, { val: {} }, { val: 2 }],
+      },
+      { inputColumnId: 'val', outputColumnId: 'output' }
+    );
+    expect(result.rows.map((row) => row.output)).toEqual([5, 12, NaN, NaN]);
+  });
+
+  it('skips undefined and null values', () => {
+    const result = runFn(
+      {
+        type: 'datatable',
+        columns: [{ id: 'val', name: 'val', meta: { type: 'number' } }],
+        rows: [
+          { val: null },
+          { val: 7 },
+          { val: undefined },
+          { val: undefined },
+          { val: undefined },
+          { val: undefined },
+          { val: '3' },
+          { val: 2 },
+          { val: null },
+        ],
+      },
+      { inputColumnId: 'val', outputColumnId: 'output' }
+    );
+    expect(result.rows.map((row) => row.output)).toEqual([0, 7, 7, 7, 7, 7, 10, 12, 12]);
+  });
+
+  it('copies over meta information from the source column', () => {
+    const result = runFn(
+      {
+        type: 'datatable',
+        columns: [
+          {
+            id: 'val',
+            name: 'val',
+            meta: {
+              type: 'number',
+
+              field: 'afield',
+              index: 'anindex',
+              params: { id: 'number', params: { pattern: '000' } },
+              source: 'synthetic',
+              sourceParams: {
+                some: 'params',
+              },
+            },
+          },
+        ],
+        rows: [{ val: 5 }],
+      },
+      { inputColumnId: 'val', outputColumnId: 'output' }
+    );
+    expect(result.columns).toContainEqual({
+      id: 'output',
+      name: 'output',
+      meta: {
+        type: 'number',
+
+        field: 'afield',
+        index: 'anindex',
+        params: { id: 'number', params: { pattern: '000' } },
+        source: 'synthetic',
+        sourceParams: {
+          some: 'params',
+        },
+      },
+    });
+  });
+
+  it('sets output name on output column if specified', () => {
+    const result = runFn(
+      {
+        type: 'datatable',
+        columns: [
+          {
+            id: 'val',
+            name: 'val',
+            meta: {
+              type: 'number',
+            },
+          },
+        ],
+        rows: [{ val: 5 }],
+      },
+      { inputColumnId: 'val', outputColumnId: 'output', outputColumnName: 'Output name' }
+    );
+    expect(result.columns).toContainEqual({
+      id: 'output',
+      name: 'Output name',
+      meta: { type: 'number' },
+    });
+  });
+
+  it('returns source table if input column does not exist', () => {
+    const input: Datatable = {
+      type: 'datatable',
+      columns: [
+        {
+          id: 'val',
+          name: 'val',
+          meta: {
+            type: 'number',
+          },
+        },
+      ],
+      rows: [{ val: 5 }],
+    };
+    expect(runFn(input, { inputColumnId: 'nonexisting', outputColumnId: 'output' })).toBe(input);
+  });
+
+  it('throws an error if output column exists already', () => {
+    expect(() =>
+      runFn(
+        {
+          type: 'datatable',
+          columns: [
+            {
+              id: 'val',
+              name: 'val',
+              meta: {
+                type: 'number',
+              },
+            },
+          ],
+          rows: [{ val: 5 }],
+        },
+        { inputColumnId: 'val', outputColumnId: 'val' }
+      )
+    ).toThrow();
+  });
+});
diff --git a/src/plugins/expressions/common/expression_functions/types.ts b/src/plugins/expressions/common/expression_functions/types.ts
index caaef541aefd5..fb1823e85b391 100644
--- a/src/plugins/expressions/common/expression_functions/types.ts
+++ b/src/plugins/expressions/common/expression_functions/types.ts
@@ -29,6 +29,7 @@ import {
   ExpressionFunctionVarSet,
   ExpressionFunctionVar,
   ExpressionFunctionTheme,
+  ExpressionFunctionCumulativeSum,
 } from './specs';
 import { ExpressionAstFunction } from '../ast';
 import { PersistableStateDefinition } from '../../../kibana_utils/common';
@@ -131,4 +132,5 @@ export interface ExpressionFunctionDefinitions {
   var_set: ExpressionFunctionVarSet;
   var: ExpressionFunctionVar;
   theme: ExpressionFunctionTheme;
+  cumulative_sum: ExpressionFunctionCumulativeSum;
 }
diff --git a/src/plugins/expressions/public/public.api.md b/src/plugins/expressions/public/public.api.md
index 95ee651d433ac..4739b9434bdaa 100644
--- a/src/plugins/expressions/public/public.api.md
+++ b/src/plugins/expressions/public/public.api.md
@@ -377,6 +377,10 @@ export interface ExpressionFunctionDefinitions {
     //
     // (undocumented)
     clog: ExpressionFunctionClog;
+    // Warning: (ae-forgotten-export) The symbol "ExpressionFunctionCumulativeSum" needs to be exported by the entry point index.d.ts
+    //
+    // (undocumented)
+    cumulative_sum: ExpressionFunctionCumulativeSum;
     // Warning: (ae-forgotten-export) The symbol "ExpressionFunctionFont" needs to be exported by the entry point index.d.ts
     //
     // (undocumented)
diff --git a/src/plugins/expressions/server/server.api.md b/src/plugins/expressions/server/server.api.md
index d5da60af8f8e5..fcdfd5ef3246c 100644
--- a/src/plugins/expressions/server/server.api.md
+++ b/src/plugins/expressions/server/server.api.md
@@ -349,6 +349,10 @@ export interface ExpressionFunctionDefinitions {
     //
     // (undocumented)
     clog: ExpressionFunctionClog;
+    // Warning: (ae-forgotten-export) The symbol "ExpressionFunctionCumulativeSum" needs to be exported by the entry point index.d.ts
+    //
+    // (undocumented)
+    cumulative_sum: ExpressionFunctionCumulativeSum;
     // Warning: (ae-forgotten-export) The symbol "ExpressionFunctionFont" needs to be exported by the entry point index.d.ts
     //
     // (undocumented)
diff --git a/x-pack/plugins/apm/public/components/app/Correlations/index.tsx b/x-pack/plugins/apm/public/components/app/Correlations/index.tsx
new file mode 100644
index 0000000000000..afee2b9f5e881
--- /dev/null
+++ b/x-pack/plugins/apm/public/components/app/Correlations/index.tsx
@@ -0,0 +1,85 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+import React from 'react';
+import url from 'url';
+import { useParams } from 'react-router-dom';
+import { useLocation } from 'react-router-dom';
+import { EuiTitle, EuiListGroup } from '@elastic/eui';
+import { useUrlParams } from '../../../hooks/useUrlParams';
+import { useApmPluginContext } from '../../../hooks/useApmPluginContext';
+
+const SESSION_STORAGE_KEY = 'apm.debug.show_correlations';
+
+export function Correlations() {
+  const location = useLocation();
+  const { serviceName } = useParams<{ serviceName?: string }>();
+  const { urlParams, uiFilters } = useUrlParams();
+  const { core } = useApmPluginContext();
+  const { transactionName, transactionType, start, end } = urlParams;
+
+  if (
+    !location.search.includes('&_show_correlations') &&
+    sessionStorage.getItem(SESSION_STORAGE_KEY) !== 'true'
+  ) {
+    return null;
+  }
+
+  sessionStorage.setItem(SESSION_STORAGE_KEY, 'true');
+
+  const query = {
+    serviceName,
+    transactionName,
+    transactionType,
+    start,
+    end,
+    uiFilters: JSON.stringify(uiFilters),
+    fieldNames:
+      'user.username,user.id,host.ip,user_agent.name,kubernetes.pod.uuid,url.domain,container.id,service.node.name',
+  };
+
+  const listItems = [
+    {
+      label: 'Show correlations between two ranges',
+      href: url.format({
+        query: {
+          ...query,
+          gap: 24,
+        },
+        pathname: core.http.basePath.prepend(`/api/apm/correlations/ranges`),
+      }),
+      isDisabled: false,
+      iconType: 'tokenRange',
+      size: 's' as const,
+    },
+
+    {
+      label: 'Show correlations for slow transactions',
+      href: url.format({
+        query: {
+          ...query,
+          durationPercentile: 95,
+        },
+        pathname: core.http.basePath.prepend(
+          `/api/apm/correlations/slow_durations`
+        ),
+      }),
+      isDisabled: false,
+      iconType: 'clock',
+      size: 's' as const,
+    },
+  ];
+
+  return (
+    <>
+      <EuiTitle size="m">
+        <h2>Correlations</h2>
+      </EuiTitle>
+
+      <EuiListGroup listItems={listItems} />
+    </>
+  );
+}
diff --git a/x-pack/plugins/apm/public/components/app/ServiceOverview/index.tsx b/x-pack/plugins/apm/public/components/app/ServiceOverview/index.tsx
index bc4a3212bafe2..fba5df5b16a4a 100644
--- a/x-pack/plugins/apm/public/components/app/ServiceOverview/index.tsx
+++ b/x-pack/plugins/apm/public/components/app/ServiceOverview/index.tsx
@@ -21,6 +21,7 @@ import { useApmPluginContext } from '../../../hooks/useApmPluginContext';
 import { MLCallout } from './ServiceList/MLCallout';
 import { useLocalStorage } from '../../../hooks/useLocalStorage';
 import { useAnomalyDetectionJobs } from '../../../hooks/useAnomalyDetectionJobs';
+import { Correlations } from '../Correlations';
 
 const initialData = {
   items: [],
@@ -117,6 +118,9 @@ export function ServiceOverview() {
   return (
     <>
       <EuiSpacer />
+
+      <Correlations />
+
       <EuiFlexGroup>
         <EuiFlexItem grow={1}>
           <LocalUIFilters {...localFiltersConfig} />
diff --git a/x-pack/plugins/apm/public/components/app/TransactionDetails/index.tsx b/x-pack/plugins/apm/public/components/app/TransactionDetails/index.tsx
index e31a2b24f1d15..b79186a90cd1d 100644
--- a/x-pack/plugins/apm/public/components/app/TransactionDetails/index.tsx
+++ b/x-pack/plugins/apm/public/components/app/TransactionDetails/index.tsx
@@ -31,6 +31,7 @@ import { fromQuery, toQuery } from '../../shared/Links/url_helpers';
 import { useUrlParams } from '../../../hooks/useUrlParams';
 import { LocalUIFilters } from '../../shared/LocalUIFilters';
 import { HeightRetainer } from '../../shared/HeightRetainer';
+import { Correlations } from '../Correlations';
 
 interface Sample {
   traceId: string;
@@ -111,6 +112,8 @@ export function TransactionDetails({
         </EuiTitle>
       </ApmHeader>
 
+      <Correlations />
+
       <EuiFlexGroup>
         <EuiFlexItem grow={1}>
           <LocalUIFilters {...localUIFiltersConfig} />
diff --git a/x-pack/plugins/apm/public/components/app/TransactionOverview/index.tsx b/x-pack/plugins/apm/public/components/app/TransactionOverview/index.tsx
index 8c7d088d36eb2..003df632d11b3 100644
--- a/x-pack/plugins/apm/public/components/app/TransactionOverview/index.tsx
+++ b/x-pack/plugins/apm/public/components/app/TransactionOverview/index.tsx
@@ -37,6 +37,7 @@ import { TransactionList } from './TransactionList';
 import { useRedirect } from './useRedirect';
 import { TRANSACTION_PAGE_LOAD } from '../../../../common/transaction_types';
 import { UserExperienceCallout } from './user_experience_callout';
+import { Correlations } from '../Correlations';
 
 function getRedirectLocation({
   urlParams,
@@ -117,6 +118,7 @@ export function TransactionOverview({ serviceName }: TransactionOverviewProps) {
 
   return (
     <>
+      <Correlations />
       <EuiSpacer />
       <EuiFlexGroup>
         <EuiFlexItem grow={1}>
diff --git a/x-pack/plugins/apm/server/lib/helpers/setup_request.ts b/x-pack/plugins/apm/server/lib/helpers/setup_request.ts
index 26896a050dd88..a8a128937fb1c 100644
--- a/x-pack/plugins/apm/server/lib/helpers/setup_request.ts
+++ b/x-pack/plugins/apm/server/lib/helpers/setup_request.ts
@@ -49,7 +49,15 @@ export interface SetupTimeRange {
 interface SetupRequestParams {
   query?: {
     _debug?: boolean;
+
+    /**
+     * Timestamp in ms since epoch
+     */
     start?: string;
+
+    /**
+     * Timestamp in ms since epoch
+     */
     end?: string;
     uiFilters?: string;
     processorEvent?: ProcessorEvent;
diff --git a/x-pack/plugins/apm/server/lib/transaction_groups/correlations/get_correlations_for_ranges.ts b/x-pack/plugins/apm/server/lib/transaction_groups/correlations/get_correlations_for_ranges.ts
new file mode 100644
index 0000000000000..3cf0271baa1c6
--- /dev/null
+++ b/x-pack/plugins/apm/server/lib/transaction_groups/correlations/get_correlations_for_ranges.ts
@@ -0,0 +1,90 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+import { rangeFilter } from '../../../../common/utils/range_filter';
+import {
+  SERVICE_NAME,
+  TRANSACTION_NAME,
+  TRANSACTION_TYPE,
+} from '../../../../common/elasticsearch_fieldnames';
+import { ProcessorEvent } from '../../../../common/processor_event';
+import { Setup, SetupTimeRange } from '../../helpers/setup_request';
+import {
+  getSignificantTermsAgg,
+  formatAggregationResponse,
+} from './get_significant_terms_agg';
+import { SignificantTermsScoring } from './scoring_rt';
+
+export async function getCorrelationsForRanges({
+  serviceName,
+  transactionType,
+  transactionName,
+  scoring,
+  gapBetweenRanges,
+  fieldNames,
+  setup,
+}: {
+  serviceName: string | undefined;
+  transactionType: string | undefined;
+  transactionName: string | undefined;
+  scoring: SignificantTermsScoring;
+  gapBetweenRanges: number;
+  fieldNames: string[];
+  setup: Setup & SetupTimeRange;
+}) {
+  const { start, end, esFilter, apmEventClient } = setup;
+
+  const baseFilters = [...esFilter];
+
+  if (serviceName) {
+    baseFilters.push({ term: { [SERVICE_NAME]: serviceName } });
+  }
+
+  if (transactionType) {
+    baseFilters.push({ term: { [TRANSACTION_TYPE]: transactionType } });
+  }
+
+  if (transactionName) {
+    baseFilters.push({ term: { [TRANSACTION_NAME]: transactionName } });
+  }
+
+  const diff = end - start + gapBetweenRanges;
+  const baseRangeStart = start - diff;
+  const baseRangeEnd = end - diff;
+  const backgroundFilters = [
+    ...baseFilters,
+    { range: rangeFilter(baseRangeStart, baseRangeEnd) },
+  ];
+
+  const params = {
+    apm: { events: [ProcessorEvent.transaction] },
+    body: {
+      size: 0,
+      query: {
+        bool: { filter: [...baseFilters, { range: rangeFilter(start, end) }] },
+      },
+      aggs: getSignificantTermsAgg({
+        fieldNames,
+        backgroundFilters,
+        backgroundIsSuperset: false,
+        scoring,
+      }),
+    },
+  };
+
+  const response = await apmEventClient.search(params);
+
+  return {
+    message: `Showing significant fields between the ranges`,
+    firstRange: `${new Date(baseRangeStart).toISOString()} - ${new Date(
+      baseRangeEnd
+    ).toISOString()}`,
+    lastRange: `${new Date(start).toISOString()} - ${new Date(
+      end
+    ).toISOString()}`,
+    response: formatAggregationResponse(response.aggregations),
+  };
+}
diff --git a/x-pack/plugins/apm/server/lib/transaction_groups/correlations/get_correlations_for_slow_transactions.ts b/x-pack/plugins/apm/server/lib/transaction_groups/correlations/get_correlations_for_slow_transactions.ts
new file mode 100644
index 0000000000000..3efc65afdfd28
--- /dev/null
+++ b/x-pack/plugins/apm/server/lib/transaction_groups/correlations/get_correlations_for_slow_transactions.ts
@@ -0,0 +1,94 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+import { asDuration } from '../../../../common/utils/formatters';
+import { ESFilter } from '../../../../typings/elasticsearch';
+import { rangeFilter } from '../../../../common/utils/range_filter';
+import {
+  SERVICE_NAME,
+  TRANSACTION_DURATION,
+  TRANSACTION_NAME,
+  TRANSACTION_TYPE,
+} from '../../../../common/elasticsearch_fieldnames';
+import { ProcessorEvent } from '../../../../common/processor_event';
+import { Setup, SetupTimeRange } from '../../helpers/setup_request';
+import { getDurationForPercentile } from './get_duration_for_percentile';
+import {
+  formatAggregationResponse,
+  getSignificantTermsAgg,
+} from './get_significant_terms_agg';
+import { SignificantTermsScoring } from './scoring_rt';
+
+export async function getCorrelationsForSlowTransactions({
+  serviceName,
+  transactionType,
+  transactionName,
+  durationPercentile,
+  fieldNames,
+  scoring,
+  setup,
+}: {
+  serviceName: string | undefined;
+  transactionType: string | undefined;
+  transactionName: string | undefined;
+  scoring: SignificantTermsScoring;
+  durationPercentile: number;
+  fieldNames: string[];
+  setup: Setup & SetupTimeRange;
+}) {
+  const { start, end, esFilter, apmEventClient } = setup;
+
+  const backgroundFilters: ESFilter[] = [
+    ...esFilter,
+    { range: rangeFilter(start, end) },
+  ];
+
+  if (serviceName) {
+    backgroundFilters.push({ term: { [SERVICE_NAME]: serviceName } });
+  }
+
+  if (transactionType) {
+    backgroundFilters.push({ term: { [TRANSACTION_TYPE]: transactionType } });
+  }
+
+  if (transactionName) {
+    backgroundFilters.push({ term: { [TRANSACTION_NAME]: transactionName } });
+  }
+
+  const durationForPercentile = await getDurationForPercentile({
+    durationPercentile,
+    backgroundFilters,
+    setup,
+  });
+
+  const params = {
+    apm: { events: [ProcessorEvent.transaction] },
+    body: {
+      size: 0,
+      query: {
+        bool: {
+          // foreground filters
+          filter: [
+            ...backgroundFilters,
+            {
+              range: { [TRANSACTION_DURATION]: { gte: durationForPercentile } },
+            },
+          ],
+        },
+      },
+      aggs: getSignificantTermsAgg({ fieldNames, backgroundFilters, scoring }),
+    },
+  };
+
+  const response = await apmEventClient.search(params);
+
+  return {
+    message: `Showing significant fields for transactions slower than ${durationPercentile}th percentile (${asDuration(
+      durationForPercentile
+    )})`,
+    response: formatAggregationResponse(response.aggregations),
+  };
+}
diff --git a/x-pack/plugins/apm/server/lib/transaction_groups/correlations/get_duration_for_percentile.ts b/x-pack/plugins/apm/server/lib/transaction_groups/correlations/get_duration_for_percentile.ts
new file mode 100644
index 0000000000000..37ee19ff40f62
--- /dev/null
+++ b/x-pack/plugins/apm/server/lib/transaction_groups/correlations/get_duration_for_percentile.ts
@@ -0,0 +1,43 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+import { ESFilter } from '../../../../typings/elasticsearch';
+import { TRANSACTION_DURATION } from '../../../../common/elasticsearch_fieldnames';
+import { ProcessorEvent } from '../../../../common/processor_event';
+import { Setup, SetupTimeRange } from '../../helpers/setup_request';
+
+export async function getDurationForPercentile({
+  durationPercentile,
+  backgroundFilters,
+  setup,
+}: {
+  durationPercentile: number;
+  backgroundFilters: ESFilter[];
+  setup: Setup & SetupTimeRange;
+}) {
+  const { apmEventClient } = setup;
+  const res = await apmEventClient.search({
+    apm: {
+      events: [ProcessorEvent.transaction],
+    },
+    body: {
+      size: 0,
+      query: {
+        bool: { filter: backgroundFilters },
+      },
+      aggs: {
+        percentile: {
+          percentiles: {
+            field: TRANSACTION_DURATION,
+            percents: [durationPercentile],
+          },
+        },
+      },
+    },
+  });
+
+  return Object.values(res.aggregations?.percentile.values || {})[0];
+}
diff --git a/x-pack/plugins/apm/server/lib/transaction_groups/correlations/get_significant_terms_agg.ts b/x-pack/plugins/apm/server/lib/transaction_groups/correlations/get_significant_terms_agg.ts
new file mode 100644
index 0000000000000..1cf0787c1d970
--- /dev/null
+++ b/x-pack/plugins/apm/server/lib/transaction_groups/correlations/get_significant_terms_agg.ts
@@ -0,0 +1,68 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+import { ESFilter } from '../../../../typings/elasticsearch';
+import { SignificantTermsScoring } from './scoring_rt';
+
+export function getSignificantTermsAgg({
+  fieldNames,
+  backgroundFilters,
+  backgroundIsSuperset = true,
+  scoring = 'percentage',
+}: {
+  fieldNames: string[];
+  backgroundFilters: ESFilter[];
+  backgroundIsSuperset?: boolean;
+  scoring: SignificantTermsScoring;
+}) {
+  return fieldNames.reduce((acc, fieldName) => {
+    return {
+      ...acc,
+      [fieldName]: {
+        significant_terms: {
+          size: 10,
+          field: fieldName,
+          background_filter: { bool: { filter: backgroundFilters } },
+
+          // indicate whether background is a superset of the foreground
+          mutual_information: { background_is_superset: backgroundIsSuperset },
+
+          // different scorings https://www.elastic.co/guide/en/elasticsearch/reference/current/search-aggregations-bucket-significantterms-aggregation.html#significantterms-aggregation-parameters
+          [scoring]: {},
+          min_doc_count: 5,
+          shard_min_doc_count: 5,
+        },
+      },
+      [`cardinality-${fieldName}`]: {
+        cardinality: { field: fieldName },
+      },
+    };
+  }, {} as Record<string, any>);
+}
+
+export function formatAggregationResponse(aggs?: Record<string, any>) {
+  if (!aggs) {
+    return;
+  }
+
+  return Object.entries(aggs).reduce((acc, [key, value]) => {
+    if (key.startsWith('cardinality-')) {
+      if (value.value > 0) {
+        const fieldName = key.slice(12);
+        acc[fieldName] = {
+          ...acc[fieldName],
+          cardinality: value.value,
+        };
+      }
+    } else if (value.buckets.length > 0) {
+      acc[key] = {
+        ...acc[key],
+        value,
+      };
+    }
+    return acc;
+  }, {} as Record<string, { cardinality: number; value: any }>);
+}
diff --git a/x-pack/plugins/apm/server/lib/transaction_groups/correlations/scoring_rt.ts b/x-pack/plugins/apm/server/lib/transaction_groups/correlations/scoring_rt.ts
new file mode 100644
index 0000000000000..cb94b6251eb07
--- /dev/null
+++ b/x-pack/plugins/apm/server/lib/transaction_groups/correlations/scoring_rt.ts
@@ -0,0 +1,16 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+import * as t from 'io-ts';
+
+export const scoringRt = t.union([
+  t.literal('jlh'),
+  t.literal('chi_square'),
+  t.literal('gnd'),
+  t.literal('percentage'),
+]);
+
+export type SignificantTermsScoring = t.TypeOf<typeof scoringRt>;
diff --git a/x-pack/plugins/apm/server/routes/correlations.ts b/x-pack/plugins/apm/server/routes/correlations.ts
new file mode 100644
index 0000000000000..5f8d2afd544f3
--- /dev/null
+++ b/x-pack/plugins/apm/server/routes/correlations.ts
@@ -0,0 +1,101 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+import * as t from 'io-ts';
+import { rangeRt } from './default_api_types';
+import { getCorrelationsForSlowTransactions } from '../lib/transaction_groups/correlations/get_correlations_for_slow_transactions';
+import { getCorrelationsForRanges } from '../lib/transaction_groups/correlations/get_correlations_for_ranges';
+import { scoringRt } from '../lib/transaction_groups/correlations/scoring_rt';
+import { createRoute } from './create_route';
+import { setupRequest } from '../lib/helpers/setup_request';
+
+export const correlationsForSlowTransactionsRoute = createRoute(() => ({
+  path: '/api/apm/correlations/slow_durations',
+  params: {
+    query: t.intersection([
+      t.partial({
+        serviceName: t.string,
+        transactionName: t.string,
+        transactionType: t.string,
+        scoring: scoringRt,
+      }),
+      t.type({
+        durationPercentile: t.string,
+        fieldNames: t.string,
+      }),
+      t.partial({ uiFilters: t.string }),
+      rangeRt,
+    ]),
+  },
+  handler: async ({ context, request }) => {
+    const setup = await setupRequest(context, request);
+    const {
+      serviceName,
+      transactionType,
+      transactionName,
+      durationPercentile,
+      fieldNames,
+      scoring = 'percentage',
+    } = context.params.query;
+
+    return getCorrelationsForSlowTransactions({
+      serviceName,
+      transactionType,
+      transactionName,
+      durationPercentile: parseInt(durationPercentile, 10),
+      fieldNames: fieldNames.split(','),
+      scoring,
+      setup,
+    });
+  },
+}));
+
+export const correlationsForRangesRoute = createRoute(() => ({
+  path: '/api/apm/correlations/ranges',
+  params: {
+    query: t.intersection([
+      t.partial({
+        serviceName: t.string,
+        transactionName: t.string,
+        transactionType: t.string,
+        scoring: scoringRt,
+        gap: t.string,
+      }),
+      t.type({
+        fieldNames: t.string,
+      }),
+      t.partial({ uiFilters: t.string }),
+      rangeRt,
+    ]),
+  },
+  handler: async ({ context, request }) => {
+    const setup = await setupRequest(context, request);
+
+    const {
+      serviceName,
+      transactionType,
+      transactionName,
+      scoring = 'percentage',
+      gap,
+      fieldNames,
+    } = context.params.query;
+
+    const gapBetweenRanges = parseInt(gap || '0', 10) * 3600 * 1000;
+    if (gapBetweenRanges < 0) {
+      throw new Error('gap must be 0 or positive');
+    }
+
+    return getCorrelationsForRanges({
+      serviceName,
+      transactionType,
+      transactionName,
+      scoring,
+      gapBetweenRanges,
+      fieldNames: fieldNames.split(','),
+      setup,
+    });
+  },
+}));
diff --git a/x-pack/plugins/apm/server/routes/create_apm_api.ts b/x-pack/plugins/apm/server/routes/create_apm_api.ts
index c1f13ee646e49..2fbe404a70d82 100644
--- a/x-pack/plugins/apm/server/routes/create_apm_api.ts
+++ b/x-pack/plugins/apm/server/routes/create_apm_api.ts
@@ -41,6 +41,10 @@ import { metricsChartsRoute } from './metrics';
 import { serviceNodesRoute } from './service_nodes';
 import { tracesRoute, tracesByIdRoute } from './traces';
 import { transactionByTraceIdRoute } from './transaction';
+import {
+  correlationsForRangesRoute,
+  correlationsForSlowTransactionsRoute,
+} from './correlations';
 import {
   transactionGroupsBreakdownRoute,
   transactionGroupsChartsRoute,
@@ -122,6 +126,10 @@ const createApmApi = () => {
     .add(listAgentConfigurationServicesRoute)
     .add(createOrUpdateAgentConfigurationRoute)
 
+    // Correlations
+    .add(correlationsForSlowTransactionsRoute)
+    .add(correlationsForRangesRoute)
+
     // APM indices
     .add(apmIndexSettingsRoute)
     .add(apmIndicesRoute)
diff --git a/x-pack/plugins/apm/typings/elasticsearch/aggregations.ts b/x-pack/plugins/apm/typings/elasticsearch/aggregations.ts
index 534321201938d..8b3163e44915a 100644
--- a/x-pack/plugins/apm/typings/elasticsearch/aggregations.ts
+++ b/x-pack/plugins/apm/typings/elasticsearch/aggregations.ts
@@ -158,6 +158,11 @@ export interface AggregationOptionsByType {
     from?: number;
     size?: number;
   };
+  significant_terms: {
+    size?: number;
+    field?: string;
+    background_filter?: Record<string, any>;
+  } & AggregationSourceOptions;
 }
 
 type AggregationType = keyof AggregationOptionsByType;
@@ -334,6 +339,17 @@ interface AggregationResponsePart<
       ? Array<{ key: number; value: number }>
       : Record<string, number>;
   };
+  significant_terms: {
+    doc_count: number;
+    bg_count: number;
+    buckets: Array<
+      {
+        bg_count: number;
+        doc_count: number;
+        key: string | number;
+      } & SubAggregationResponseOf<TAggregationOptionsMap['aggs'], TDocument>
+    >;
+  };
   bucket_sort: undefined;
 }
 
diff --git a/x-pack/test/apm_api_integration/basic/tests/correlations/ranges.ts b/x-pack/test/apm_api_integration/basic/tests/correlations/ranges.ts
new file mode 100644
index 0000000000000..f013520fa163b
--- /dev/null
+++ b/x-pack/test/apm_api_integration/basic/tests/correlations/ranges.ts
@@ -0,0 +1,96 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+import expect from '@kbn/expect';
+import { format } from 'url';
+import { expectSnapshot } from '../../../common/match_snapshot';
+import { PromiseReturnType } from '../../../../../plugins/apm/typings/common';
+import { FtrProviderContext } from '../../../common/ftr_provider_context';
+import archives_metadata from '../../../common/archives_metadata';
+
+export default function ApiTest({ getService }: FtrProviderContext) {
+  const supertest = getService('supertest');
+  const esArchiver = getService('esArchiver');
+  const archiveName = 'apm_8.0.0';
+  const range = archives_metadata[archiveName];
+
+  // url parameters
+  const start = '2020-09-29T14:45:00.000Z';
+  const end = range.end;
+  const fieldNames =
+    'user.username,user.id,host.ip,user_agent.name,kubernetes.pod.uuid,url.domain,container.id,service.node.name';
+
+  describe('Ranges', () => {
+    const url = format({
+      pathname: `/api/apm/correlations/ranges`,
+      query: { start, end, fieldNames },
+    });
+
+    describe('when data is not loaded ', () => {
+      it('handles the empty state', async () => {
+        const response = await supertest.get(url);
+        expect(response.status).to.be(200);
+        expect(response.body.response).to.be(undefined);
+      });
+    });
+
+    describe('when data is loaded', () => {
+      let response: PromiseReturnType<typeof supertest.get>;
+      before(async () => {
+        await esArchiver.load(archiveName);
+        response = await supertest.get(url);
+      });
+
+      after(() => esArchiver.unload(archiveName));
+
+      it('returns successfully', () => {
+        expect(response.status).to.eql(200);
+      });
+
+      it('returns fields in response', () => {
+        expectSnapshot(Object.keys(response.body.response)).toMatchInline(`
+          Array [
+            "service.node.name",
+            "host.ip",
+            "user.id",
+            "user_agent.name",
+            "container.id",
+            "url.domain",
+          ]
+        `);
+      });
+
+      it('returns cardinality for each field', () => {
+        const cardinalitys = Object.values(response.body.response).map(
+          (field: any) => field.cardinality
+        );
+
+        expectSnapshot(cardinalitys).toMatchInline(`
+          Array [
+            5,
+            6,
+            20,
+            6,
+            5,
+            4,
+          ]
+        `);
+      });
+
+      it('returns buckets', () => {
+        const { buckets } = response.body.response['user.id'].value;
+        expectSnapshot(buckets[0]).toMatchInline(`
+          Object {
+            "bg_count": 2,
+            "doc_count": 7,
+            "key": "20",
+            "score": 3.5,
+          }
+        `);
+      });
+    });
+  });
+}
diff --git a/x-pack/test/apm_api_integration/basic/tests/correlations/slow_durations.ts b/x-pack/test/apm_api_integration/basic/tests/correlations/slow_durations.ts
new file mode 100644
index 0000000000000..78dca5100dece
--- /dev/null
+++ b/x-pack/test/apm_api_integration/basic/tests/correlations/slow_durations.ts
@@ -0,0 +1,138 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+import expect from '@kbn/expect';
+import { format } from 'url';
+import { PromiseReturnType } from '../../../../../plugins/apm/typings/common';
+import { FtrProviderContext } from '../../../common/ftr_provider_context';
+import { expectSnapshot } from '../../../common/match_snapshot';
+import archives_metadata from '../../../common/archives_metadata';
+
+export default function ApiTest({ getService }: FtrProviderContext) {
+  const supertest = getService('supertest');
+  const esArchiver = getService('esArchiver');
+  const archiveName = 'apm_8.0.0';
+  const range = archives_metadata[archiveName];
+
+  // url parameters
+  const start = range.start;
+  const end = range.end;
+  const durationPercentile = 95;
+  const fieldNames =
+    'user.username,user.id,host.ip,user_agent.name,kubernetes.pod.uuid,url.domain,container.id,service.node.name';
+
+  describe('Slow durations', () => {
+    const url = format({
+      pathname: `/api/apm/correlations/slow_durations`,
+      query: { start, end, durationPercentile, fieldNames },
+    });
+
+    describe('when data is not loaded ', () => {
+      it('handles the empty state', async () => {
+        const response = await supertest.get(url);
+
+        expect(response.status).to.be(200);
+        expect(response.body.response).to.be(undefined);
+      });
+    });
+
+    describe('with default scoring', () => {
+      let response: PromiseReturnType<typeof supertest.get>;
+      before(async () => {
+        await esArchiver.load(archiveName);
+        response = await supertest.get(url);
+      });
+
+      after(() => esArchiver.unload(archiveName));
+
+      it('returns successfully', () => {
+        expect(response.status).to.eql(200);
+      });
+
+      it('returns fields in response', () => {
+        expectSnapshot(Object.keys(response.body.response)).toMatchInline(`
+          Array [
+            "service.node.name",
+            "host.ip",
+            "user.id",
+            "user_agent.name",
+            "container.id",
+            "url.domain",
+          ]
+        `);
+      });
+
+      it('returns cardinality for each field', () => {
+        const cardinalitys = Object.values(response.body.response).map(
+          (field: any) => field.cardinality
+        );
+
+        expectSnapshot(cardinalitys).toMatchInline(`
+          Array [
+            5,
+            6,
+            3,
+            5,
+            5,
+            4,
+          ]
+        `);
+      });
+
+      it('returns buckets', () => {
+        const { buckets } = response.body.response['user.id'].value;
+        expectSnapshot(buckets[0]).toMatchInline(`
+          Object {
+            "bg_count": 32,
+            "doc_count": 6,
+            "key": "2",
+            "score": 0.1875,
+          }
+        `);
+      });
+    });
+
+    describe('with different scoring', () => {
+      before(async () => esArchiver.load(archiveName));
+      after(() => esArchiver.unload(archiveName));
+
+      it(`returns buckets for each score`, async () => {
+        const promises = ['percentage', 'jlh', 'chi_square', 'gnd'].map(async (scoring) => {
+          const response = await supertest.get(
+            format({
+              pathname: `/api/apm/correlations/slow_durations`,
+              query: { start, end, durationPercentile, fieldNames, scoring },
+            })
+          );
+
+          return { name: scoring, value: response.body.response['user.id'].value.buckets[0].score };
+        });
+
+        const res = await Promise.all(promises);
+        expectSnapshot(res).toMatchInline(`
+          Array [
+            Object {
+              "name": "percentage",
+              "value": 0.1875,
+            },
+            Object {
+              "name": "jlh",
+              "value": 3.33506905769659,
+            },
+            Object {
+              "name": "chi_square",
+              "value": 219.192006524483,
+            },
+            Object {
+              "name": "gnd",
+              "value": 0.671406580688819,
+            },
+          ]
+        `);
+      });
+    });
+  });
+}
diff --git a/x-pack/test/apm_api_integration/basic/tests/index.ts b/x-pack/test/apm_api_integration/basic/tests/index.ts
index 19dd82d617bd9..df3e60d79aca5 100644
--- a/x-pack/test/apm_api_integration/basic/tests/index.ts
+++ b/x-pack/test/apm_api_integration/basic/tests/index.ts
@@ -56,5 +56,10 @@ export default function apmApiIntegrationTests({ loadTestFile }: FtrProviderCont
     describe('Metrics', function () {
       loadTestFile(require.resolve('./metrics_charts/metrics_charts'));
     });
+
+    describe('Correlations', function () {
+      loadTestFile(require.resolve('./correlations/slow_durations'));
+      loadTestFile(require.resolve('./correlations/ranges'));
+    });
   });
 }