> ## Documentation Index
> Fetch the complete documentation index at: https://doc.lucidworks.com/llms.txt
> Use this file to discover all available pages before exploring further.

# Aggregation Jobs

export const schema = {
  "type": "object",
  "title": "SQL Aggregation",
  "description": "Aggregates signals or document data using SQL and writes the results to an output collection.",
  "required": ["id", "inputCollection", "sql", "dataFormat", "type"],
  "properties": {
    "id": {
      "type": "string",
      "title": "Spark Job ID",
      "description": "The ID for this Spark job. Used in the API to reference this job. Allowed characters: a-z, A-Z, dash (-) and underscore (_).",
      "maxLength": 63,
      "pattern": "[a-zA-Z][_\\-a-zA-Z0-9]*[a-zA-Z0-9]?"
    },
    "sparkConfig": {
      "type": "array",
      "title": "Spark Settings",
      "description": "Spark configuration settings.",
      "hints": ["advanced"],
      "items": {
        "type": "object",
        "required": ["key"],
        "properties": {
          "key": {
            "type": "string",
            "title": "Parameter Name"
          },
          "value": {
            "type": "string",
            "title": "Parameter Value"
          }
        }
      }
    },
    "inputCollection": {
      "type": "string",
      "title": "Source Collection",
      "description": "Specifies the collection containing signals to aggregate."
    },
    "outputCollection": {
      "type": "string",
      "title": "Output Collection",
      "description": "Specifies the collection where aggregation results are written. Required when the output destination is a separate collection.",
      "hints": ["advanced"],
      "minLength": 1
    },
    "rows": {
      "type": "integer",
      "title": "Batch Size",
      "description": "Sets the number of rows read from the source collection per request.",
      "default": 10000,
      "hints": ["advanced"]
    },
    "sql": {
      "type": "string",
      "title": "SQL",
      "description": "Specifies the SQL statement that performs the aggregation. Omit time range filtering from the WHERE clause. It is applied automatically.",
      "hints": ["lengthy", "code/sql"],
      "minLength": 1
    },
    "rollupSql": {
      "type": "string",
      "title": "Rollup SQL",
      "description": "Specifies an optional SQL statement for rolling up previously aggregated documents. If blank, the aggregation framework handles rollup automatically.",
      "hints": ["lengthy", "code/sql", "advanced"],
      "minLength": 1
    },
    "readOptions": {
      "type": "array",
      "title": "Read Options",
      "description": "Sets additional key-value options passed to the Spark reader when loading input from Solr or other sources.",
      "hints": ["advanced"],
      "items": {
        "type": "object",
        "required": ["key"],
        "properties": {
          "key": {
            "type": "string",
            "title": "Parameter Name"
          },
          "value": {
            "type": "string",
            "title": "Parameter Value"
          }
        }
      }
    },
    "sourceCatchup": {
      "type": "boolean",
      "title": "Aggregate New and Merge with Existing",
      "description": "When enabled, processes only signals created since the last successful job run instead of the full dataset.",
      "default": true,
      "hints": ["advanced"]
    },
    "sourceRemove": {
      "type": "boolean",
      "title": "Remove signals from source",
      "description": "When enabled, removes processed signals from the source collection after aggregation completes.",
      "default": false,
      "hints": ["advanced"]
    },
    "aggregationTime": {
      "type": "string",
      "title": "Aggregation Time",
      "description": "Specifies the timestamp used for aggregation results. Defaults to `NOW`.",
      "hints": ["advanced"],
      "format": "date-time"
    },
    "referenceTime": {
      "type": "string",
      "title": "Reference Time",
      "description": "Specifies the timestamp used as the reference point for decay calculations and the value of `NOW`.",
      "hints": ["advanced"],
      "format": "date-time"
    },
    "skipCheckEnabled": {
      "type": "boolean",
      "title": "Job Skip Check Enabled?",
      "description": "When catch-up is enabled, runs a fast Solr query to determine if new data exists before starting aggregation.",
      "default": true,
      "hints": ["advanced"]
    },
    "skipJobIfSignalsEmpty": {
      "type": "boolean",
      "title": "Skip Job run",
      "description": "When enabled, skips the job run if the signals collection is empty.",
      "hints": ["advanced"]
    },
    "parameters": {
      "type": "array",
      "title": "Parameters",
      "description": "Specifies key-value parameters that control aggregation behavior.",
      "hints": ["advanced"],
      "items": {
        "type": "object",
        "required": ["key"],
        "properties": {
          "key": {
            "type": "string",
            "title": "Parameter Name"
          },
          "value": {
            "type": "string",
            "title": "Parameter Value"
          }
        }
      }
    },
    "signalTypes": {
      "type": "array",
      "title": "Signal Types",
      "description": "Specifies the signal event types to include in aggregation. If not set, all signal types are included.",
      "items": {
        "type": "string"
      }
    },
    "selectQuery": {
      "type": "string",
      "title": "Query",
      "description": "Specifies the query used to select input documents for aggregation.",
      "default": "*:*",
      "hints": ["advanced"],
      "minLength": 1
    },
    "timeRange": {
      "type": "string",
      "title": "Time Range",
      "description": "Specifies the time range used to filter signals.",
      "hints": ["advanced"],
      "minLength": 1
    },
    "useNaturalKey": {
      "type": "boolean",
      "title": "Use Natural Key?",
      "description": "When enabled, uses a natural key from the raw signals data for aggregation instead of Solr-generated UUIDs.",
      "default": true,
      "hints": ["advanced"]
    },
    "optimizeSegments": {
      "type": "integer",
      "title": "Optimize Segments",
      "description": "Optimizes the output Solr collection to the specified number of segments after aggregation. Set to `0` to skip optimization.",
      "default": 0,
      "hints": ["advanced"],
      "minimum": 0,
      "exclusiveMinimum": false
    },
    "dataFormat": {
      "type": "string",
      "title": "Data format",
      "description": "Specifies the Spark-compatible input data format, such as `solr`, `parquet`, or `orc`.",
      "default": "solr",
      "minLength": 1
    },
    "sparkSQL": {
      "type": "string",
      "title": "Spark SQL filter query",
      "description": "Filters the loaded DataFrame using a Spark SQL query before processing. The input data is registered as a temporary table named `spark_input`.",
      "default": "SELECT * from spark_input",
      "hints": ["code/sql", "advanced"]
    },
    "sparkPartitions": {
      "type": "integer",
      "title": "Set minimum Spark partitions for input",
      "description": "Sets the number of Spark partitions to repartition input into before processing. Increase for greater parallelism on large datasets.",
      "default": 200,
      "hints": ["advanced"]
    },
    "type": {
      "type": "string",
      "title": "Spark Job Type",
      "enum": ["aggregation"],
      "default": "aggregation",
      "hints": ["readonly"]
    }
  },
  "additionalProperties": true,
  "category": "Other",
  "categoryPriority": 1
};

export const SchemaParamFields = ({schema}) => {
  const sanitize = str => {
    if (typeof str !== "string") return str;
    return str.replace(/^"(.*)"$/s, "$1").replace(/\\/g, "").replace(/"/g, "'");
  };
  const renderMd = str => {
    const s = sanitize(str);
    const text = (/[.!?]\)*$/).test(s) ? s : `${s}.`;
    return text.split(/(\*\*[^*]+\*\*|_[^_]+_|`[^`]+`)/g).map((part, i) => {
      if (part.startsWith("**")) return <strong key={i}>{part.slice(2, -2)}</strong>;
      if (part.startsWith("_")) return <em key={i}>{part.slice(1, -1)}</em>;
      if (part.startsWith("`")) return <code key={i}>{part.slice(1, -1)}</code>;
      return part;
    });
  };
  const {description, properties = {}, required: requiredProps = []} = schema;
  const visibleProps = useMemo(() => Object.entries(properties).filter(([, prop]) => !prop.hints?.includes("hidden")), [properties]);
  const renderProp = ([name, prop]) => {
    const isRequired = requiredProps.includes(name);
    const hasDefault = prop.default !== undefined;
    const rawDefault = prop.default;
    const hints = prop.hints || [];
    const isComplexDefault = hasDefault && (typeof rawDefault === "object" || typeof rawDefault === "string" && (rawDefault.length > 20 || rawDefault.includes('"')));
    const postBadges = [];
    if (prop.title) {
      postBadges.push(<><span className="text-stone-400 dark:text-stone-500">API property: </span>{name}</>);
    }
    const constraints = [];
    if (prop.minimum !== undefined && prop.maximum !== undefined) {
      constraints.push(`Range: ${prop.minimum} – ${prop.maximum}`);
    } else if (prop.minimum !== undefined) {
      constraints.push(`Min: ${prop.minimum}`);
    } else if (prop.maximum !== undefined) {
      constraints.push(`Max: ${prop.maximum}`);
    }
    if (prop.minLength !== undefined && prop.maxLength !== undefined) {
      constraints.push(`Length: ${prop.minLength} – ${prop.maxLength}`);
    } else if (prop.minLength !== undefined) {
      constraints.push(`Min length: ${prop.minLength}`);
    } else if (prop.maxLength !== undefined) {
      constraints.push(`Max length: ${prop.maxLength}`);
    }
    const fieldProps = {
      key: name,
      body: prop.title || name,
      type: prop.type,
      ...postBadges.length > 0 && ({
        post: postBadges
      }),
      ...isRequired && ({
        required: true
      }),
      ...!isComplexDefault && hasDefault ? {
        default: sanitize(String(rawDefault))
      } : {}
    };
    const isObject = prop.type === "object" && prop.properties;
    const isArrayOfObjects = prop.type === "array" && prop.items?.type === "object" && prop.items.properties;
    return <ParamField {...fieldProps}>
        {prop.description && <p>{renderMd(prop.description)}</p>}

        {prop.enum && <p>
            Allowed values: 
            {prop.enum.map((v, i) => <>{i > 0 && ", "}<code key={i}>{String(v)}</code></>)}
          </p>}

        {constraints.length > 0 && <p className="text-stone-500 dark:text-stone-400 text-sm">
            {constraints.join(" · ")}
          </p>}

        {isComplexDefault && <div className="flex">
            <p>
              <strong>Default:</strong>
            </p>
            <pre className="!my-0">
              <code>
                {JSON.stringify(rawDefault, null, 2)}
              </code>
            </pre>
          </div>}

        {isArrayOfObjects && <Expandable title="item properties">
            <SchemaParamFields schema={{
      properties: prop.items.properties,
      required: prop.items.required
    }} />
          </Expandable>}

        {isObject && <Expandable title="properties">
            <SchemaParamFields schema={{
      properties: prop.properties,
      required: prop.required
    }} />
          </Expandable>}
      </ParamField>;
  };
  return <div>
      {description && <p>{renderMd(description)}</p>}

      {visibleProps.map(renderProp)}
    </div>;
};

export const LwTemplate = ({title = "Key questions to get you started", icon = "sparkles", cta = "Powered by Agent Studio", linkHref = "https://lucidworks.com/demo/?utm_source=docs&utm_medium=referral&utm_campaign=docs_cta_ai"}) => {
  const [isLoaded, setIsLoaded] = useState(false);
  useEffect(() => {
    const timer = setTimeout(() => {
      setIsLoaded(true);
    }, 500);
    return () => clearTimeout(timer);
  }, []);
  return <div className="lw-template-container">
      <Card title={title} icon={icon}>
        {isLoaded && <span dangerouslySetInnerHTML={{
    __html: `<lw-template id="a029c1a9-28be-427e-b0e1-5d918920246a"></lw-template
            >`
  }} />}
        <Link href={linkHref} className="agent-studio-link text-left text-gray-600 gap-2 dark:text-gray-400 text-sm font-medium flex flex-row items-center hover:text-primary dark:hover:text-primary-light group-hover:text-primary group-hover:dark:text-primary-light">Powered by Lucidworks Agent Studio</Link>
      </Card>
    </div>;
};

[localhost link]: http://localhost:3000/docs/5/fusion/reference/config-ref/jobs/aggregations/overview

[mintlify link]: https://doc.lucidworks.com/docs/5/fusion/reference/config-ref/jobs/aggregations/overview

[old doc.lw link]: https://doc.lucidworks.com/fusion/5.9/590

Aggregation jobs compile your raw signals into aggregated signals. Most Fusion jobs that consume signals require aggregated signals.

|                      |                                                                                                                                                                      |
| -------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| **Default job name** | `COLLECTION_NAME_click_signals_aggregation`                                                                                                                          |
| **Input**            | Raw signals (the `COLLECTION_NAME_signals` collection by default)                                                                                                    |
| **Output**           | [Aggregated signals](/docs/5/fusion/reference/config-ref/jobs/aggregations/built-in-sql-aggregation-jobs) (the `COLLECTION_NAME_signals_aggr` collection by default) |

|                          | query | count\_i | type | timestamp\_tdt | user\_id | doc\_id | session\_id | fusion\_query\_id |
| ------------------------ | ----- | -------- | ---- | -------------- | -------- | ------- | ----------- | ----------------- |
| Required signals fields: | ✅     | ✅        | ✅    | ✅              |          | ✅       |             | See note below.   |

**Note:** Required if you are using response signals.

<LwTemplate />

## Aggregation properties

The aggregation process is specified by an aggregation type consisting of the following list of properties:

| Name               | Description                                                            |
| ------------------ | ---------------------------------------------------------------------- |
| `id`               | Aggregation ID                                                         |
| `groupingFields`   | List of signal field names                                             |
| `signalTypes`      | List of signal types                                                   |
| `aggregator`       | Symbolic name of the aggregator implementation                         |
| `selectQuery`      | Query string, default `*:*`                                            |
| `sort`             | Ordering of aggregated signals                                         |
| `timeRange`        | String specifying time range, e.g., `[* TO NOW]`                       |
| `outputPipeline`   | Pipeline ID for processing aggregated events                           |
| `outputCollection` | Output collection name                                                 |
| `rollupPipeline`   | Rollup pipeline ID                                                     |
| `rollupAggregator` | Name of the aggregator implementation used for rollups                 |
| `sourceRemove`     | Boolean, default is false                                              |
| `sourceCatchup`    | Boolean, default is true                                               |
| `outputRollup`     | Boolean, default is true                                               |
| `aggregates`       | List of aggregation functions                                          |
| `params`           | Arbitrary parameters to be used by specific aggregator implementations |

## Configuration properties

<SchemaParamFields schema={schema} />