Skip to main content

Aggregate to earliest or latest value

When you ingest data into Imply Polaris, you can aggregate, or roll up, individual records to store as a single record. For a string or numeric input field, you can use an input expression to store the input field's value that corresponds to the earliest or latest timestamp for the set of aggregated rows. The ingested field populates a measure in an aggregate table.

This topic shows you how to ingest the earliest or latest value of an input field into a table.

Function reference​

Use the following functions for input expressions in ingestion jobs:

EARLIEST_BY(expr, timestampExpr)
LATEST_BY(expr, timestampExpr)

Each function takes the following parameters:

  • expr: string or numeric input field to aggregate
  • timestampExpr: timestamp expression to reference for the earliest or latest time
    To generate the timestamp expression:
    • If you have a string input field with ISO 8601 timestamps, use the function TIME_PARSE(expr).
    • If you have a long input field with milliseconds since Unix epoch timestamps, use the function MILLIS_TO_TIMESTAMP(expr).

For more details on each function, including how to control the number of bytes allowed for aggregation, see Druid SQL aggregation functions.

info

If you create the ingestion job using the load data wizard UI, the sampling preview may not reflect the correct timestampExpr, but ingestion still works correctly.

Stored aggregation​

Polaris stores the aggregation as a JSON object that resembles the following:

{
  "lhs": 1704070018771,
  "rhs": 5
}

The lhs value stores the reference time in milliseconds since Unix epoch, and rhs stores the aggregated data value.

When you query the aggregated column with EARLIEST or LATEST, Polaris returns the value in rhs.

Workflow​

Use the following process to ingest and query earliest or latest values:

  1. Create an aggregate table. When you use the flexible schema mode (recommended), you don't need to define the column since Polaris automatically assigns the correct type at ingestion.

    Strict schema enforcement

    If you need to enforce a strict schema, declare the column before ingestion using the data type Long string pair (longStringPair using the Tables API).

    This option works well when you aggregate a string input (expr is a string type). To aggregate a numeric input, you must use flexible mode otherwise Polaris doesn't resolve to the correct aggregated value.

    If you already have an aggregate table in strict mode, you can convert it to flexible. You still need to update or recreate any existing ingestion job to define the new column mapping, described in the next step.

  2. Create an ingestion job. For the aggregated column, define its SQL expression with the relevant aggregation function. For example, EARLIEST_BY("score", TIME_PARSE("event_time")).

  3. Start the ingestion job.

  4. Query the aggregated column using EARLIEST(expr) or LATEST(expr), where expr is the aggregated column. Polaris determines the earliest or latest value based on the reference timestamp timestampExpr you specified at ingestion.

    Don't use EARLIEST_BY or LATEST_BY on the aggregated column at query time. Polaris disregards timestampExpr since the column already contains the reference timestamp from ingestion.

    info

    You can't use EARLIEST or LATEST to aggregate at ingestion.

Example​

Consider the following example data representing transactions on an e-commerce platform:

{ "received_at": "2024-06-15T10:05:00Z", "order_updated_at": "2024-06-15T09:58:12Z", "order_id": "ORD-8001", "customer_id": "C123", "status": "placed", "amount": 120.00, "order_version": 1 }
{ "received_at": "2024-06-15T10:31:00Z", "order_updated_at": "2024-06-15T10:28:45Z", "order_id": "ORD-8001", "customer_id": "C123", "status": "processing", "amount": 120.00, "order_version": 2 }
{ "received_at": "2024-06-15T14:08:00Z", "order_updated_at": "2024-06-15T13:55:30Z", "order_id": "ORD-8002", "customer_id": "C146", "status": "placed", "amount": 45.50, "order_version": 1 }
{ "received_at": "2024-06-16T14:02:00Z", "order_updated_at": "2024-06-15T13:55:30Z", "order_id": "ORD-8001", "customer_id": "C123", "status": "shipped", "amount": 120.00, "order_version": 3 }

The data shows two orders, ORD-8001 and ORD-8002. The records track each stage of the order—when it was placed, processed, and shipped. You want to ingest the data and aggregate the results to unique orders received per day. Instead of retaining the detailed stages of each order, you just want to know the latest order version and its status.

Ingest data​

To ingest the data:

  1. Prepare the sample data:
    1. Download the sample input data, orders.json.
    2. From the left navigation menu, go to Sources.
    3. Click the Create source > Select files from computer, and upload orders.json.
  2. Create an aggregate table in flexible mode:
    1. Click Tables from the left navigation menu of the Polaris UI.
    2. Click Create table.
    3. Name the table store_orders. Select the Aggregate table type and Flexible schema mode. Click Next.
  3. Start ingestion:
    1. From the table view, click Load data > Insert data.
    2. On the Insert data step, click Switch to code editor > SQL.
    3. Enter the following statement, then click Start ingestion.
INSERT INTO "store_orders"
SELECT
  TIME_FLOOR(TIME_PARSE("received_at"), 'P1D', NULL, 'UTC') AS "__time",
  "order_id" AS "order_id",
  "customer_id" AS "customer_id",
  "amount" AS "amount",
  COUNT(*) AS "__count",
  LATEST_BY("status", TIME_PARSE("order_updated_at")) AS "last_status",
  LATEST_BY("order_version", TIME_PARSE("order_updated_at")) AS "last_version"
FROM TABLE(POLARIS_SOURCE('{"fileList":["orders.json"],"inputSchema":[{"dataType":"string","name":"order_updated_at"},{"dataType":"string","name":"received_at"},{"dataType":"string","name":"order_id"},{"dataType":"string","name":"customer_id"},{"dataType":"string","name":"status"},{"dataType":"double","name":"amount"},{"dataType":"long","name":"order_version"}],"formatSettings":{"format":"nd-json"},"type":"uploaded"}'))
GROUP BY 1, 2, 3, 4
PARTITIONED BY DAY

Notice the following aspects in the SQL statement:

  • TIME_FLOOR with P1D aggregates the results by day based on received_at timestamps. This action rolls up rows that have identical order_id, customer_id, and amount.
  • For the rows rolled up, LATEST_BY for status and version stores the respective latest values based on order_updated_at timestamps.

The resulting table contains three rows of data: one for ORD-8001 on 2024-06-15, one for ORD-8002 on the same day, and another one for ORD-8001 on the next day. In a realistic scenario, the time lapsed between order_updated_at and received_at might represent pipeline latency for when the e-commerce platform recorded the change and when the event was collected and stored.

View and query data​

  1. View the store_orders table. It should resemble the following:

    Polaris aggregated table

  2. Select the first cell in the last_status column. Note the aggregated value:

    {"lhs":1718447325000,"rhs":"processing"}

    Polaris aggregated cell

  3. In the table view, click Query > SQL workbench.

  4. Run the following query to view the last status and version for each order:

    SELECT
      order_id,
      LATEST(last_status) as last_status,
      LATEST(last_version) as last_version
    FROM "store_orders"
    GROUP BY 1

    You should observe the results:

    "order_id","last_status","last_version"
    "ORD-8001","shipped","3"
    "ORD-8002","placed","1"

    Notice that the SQL uses the LATEST function in the query. The latest value was determined from order_updated_at. Any earliest or latest evaluations on columns aggregated at ingestion use the reference timestamp from ingestion.

  5. You can also evaluate earliest and latest values at query time, operating on columns that weren't aggregated at ingestion time. Run the following query to determine what order was recorded last on each day:

    SELECT
      "__time",
      LATEST_BY("order_id", "__time") as globalLastId
    FROM "store_orders"
    GROUP BY 1

    You should observe the result:

     "__time","globalLastId"
     "2024-06-15T00:00:00.000Z","ORD-8002"
     "2024-06-16T00:00:00.000Z","ORD-8001"

    Notice that LATEST_BY operates on order_id, a non-aggregated dimension, with respect to __time (received_at). If you were to call LATEST_BY on last_status, the underlying evaluation would be fixed on the timestamp from order_updated_at.

Learn more​

For more information, see the following topics: