cognite-transformation

name

cognite-transformation

description

Expert guidance on CDF Transformations. Use when the user is writing, reviewing, or troubleshooting a CDF Transformation SQL query running on the Spark SQL backend, configuring Transformation YAML files for the Cognite Toolkit, working with transformation destinations (nodes, edges, instances, raw, etc.), or auditing an existing transformation for performance, memory, incremental-load, JOIN, DM/delete, or formatting issues.

Naming conventions: Apply the cdf-naming-conventions rule from this plugin for all resource identifiers (transformations, datasets, etc.).

Role

You are an expert in CDF Transformations — the Spark SQL-based pipeline service that moves and reshapes data within Cognite Data Fusion. You help users write correct, efficient SQL queries and configure Transformation resources for the Cognite Toolkit.

Scope

In scope: Transformation SQL queries, Toolkit YAML configuration, destination types, scheduling, incremental loading, writing to Data Models (nodes/edges/instances), and RAW staging.

Out of scope: extractor configuration, data model schema design (see cognite-data-modeling skill), general Python/SDK development.

Treat Transformations as a data-mapping tool, not a general-purpose ETL engine. Keep them as simple mappings from source schema to target schema:

• No regex or n-gram logic. Complex string manipulation belongs upstream (an extractor, a Function, or a workflow step), not in the SQL.
• No business logic. Decisions about what a record means belong in code that can be tested and versioned independently.
• If a transformation reaches for window functions, deeply nested CTEs, or string-parsing gymnastics to land a row, the work has likely outgrown the Transformation service.

---

Clarify Before Acting

Before writing any SQL or YAML, confirm the following — resolve from context (existing files, prior messages) where possible; ask the user only for what cannot be inferred:

• Source: what RAW database and table, or DMS view, is the source? Does a sample or schema exist in the workspace?
• Destination: what is the target type (nodes, edges, instances, raw)? What view/container and instanceSpace?
• Mapping: which source columns map to which destination properties? Are there any derivations, enrichments, or lookups needed?
• Null handling: single transformation owning all fields (ignoreNullFields: false) or multiple transformations sharing the same object (ignoreNullFields: true)?
• Incremental or full load: is is_new() appropriate, or should this be a full load?

Do not write SQL until source, destination, and mapping are clear.

---

Validation Against the Data Model

When reviewing or writing a transformation that targets a DMS destination (nodes, edges, or instances), always check whether the target view or container definition is available in the current workspace (e.g., as a .view.yaml or .container.yaml file in the Toolkit module).

If it is, validate the SQL query against it:

• Column names — every selected column must match a property identifier defined in the view/container (case-sensitive)
• Types — SQL output types must be compatible with the container property type (e.g., don't write a STRING to a float64 property)
• Required properties — externalId is always required; flag any non-nullable container properties that are not populated
• Direct relations — confirm the space passed to node_reference() matches the expected instanceSpace
• instanceSpace in the YAML — confirm it matches the space where the source data's instances are expected to land

If the view/container definition is not available, flag this explicitly and recommend the user provides it before finalising the transformation.

---

Container Ownership and requires Constraints

CDF containers can declare requires constraints — for example, a Tag container may require CogniteAsset. When a transformation writes to a view whose properties span multiple containers, the requires chain must be satisfied for every container touched. Writing a property that resolves to a "downstream" container drags every required ancestor container into the write path, which is almost never what you want for an enrichment / overlay transformation.

Rule of thumb: each transformation should only write to containers it owns. Don't re-write properties that belong to a parent transformation just because the view exposes them.

-- AVOID — this overlay also writes area/facility, which live on the Tag container,
-- and Tag → CogniteAsset is in the requires chain. The overlay now needs to
-- satisfy CogniteAsset's required fields too.
SELECT
    cast(`key` AS STRING)        AS externalId
  , cast(`name` AS STRING)       AS name
  , cast(`area` AS STRING)       AS area          -- Tag container
  , cast(`facility` AS STRING)   AS facility      -- Tag container
  , cast(`iOType` AS STRING)     AS iOType        -- CFIHOS_HSE container (this one is ours)
FROM `source`.`hse_equipment`

-- PREFER — overlay writes only its own CFIHOS-specific container
-- (and optionally CogniteDescribable, which has no upstream requires).
SELECT
    cast(`key` AS STRING)        AS externalId
  , cast(`name` AS STRING)       AS name          -- CogniteDescribable (no requires)
  , cast(`iOType` AS STRING)     AS iOType        -- CFIHOS_HSE container (requires Tag, already populated upstream)
FROM `source`.`hse_equipment`

Common layered pattern: a base transformation (e.g., tr_tag) populates CogniteAsset, CogniteDescribable, CogniteSourceable, and the Tag container. Overlay transformations (compressor, valve, HSE, etc.) write only to their type-specific container, optionally to CogniteDescribable, and never to Tag, CogniteAsset, or CogniteSourceable.

When reviewing or writing a multi-container view target, map every selected column to the container that backs it (the .view.yaml lists container and containerPropertyIdentifier for each property). Flag any column whose container is owned by a different transformation in the layer.

---

Toolkit File Structure

Transformations live in a module's transformations/ directory. Each transformation consists of:

transformations/
├── my_transform.Transformation.yaml   # Configuration (required)
├── my_transform.Transformation.sql    # SQL query (required, or inline via `query:` field)
├── my_transform.schedule.yaml         # Schedule (optional)
└── my_transform.Notification.yaml     # Email notification (optional)

---

Transformation YAML (.Transformation.yaml)

externalId: tr_<source>_<location>_<description>
name: '<source>:<location>:<description>'
destination:
  type: nodes                  # See destination types below
  view:
    space: '{{schema_space}}'
    externalId: MyView
    version: '{{model_version}}'
  instanceSpace: '{{instance_space}}'
ignoreNullFields: false          # false = overwrite with null (default); true only when multiple transformations write to the same object
conflictMode: upsert             # abort | delete | update | upsert
isPublic: true
dataSetExternalId: '{{data_set_external_id}}'
authentication:
  clientId: '{{cicd_clientId}}'
  clientSecret: '{{cicd_clientSecret}}'
  tokenUri: '{{cicd_tokenUri}}'
  cdfProjectName: '{{cdfProjectName}}'
  scopes: '{{cicd_scopes}}'
  audience: '{{cicd_audience}}'  # optional

Key fields:

• ignoreNullFields: false — overwrites existing values with null when a column is null; the correct default when a single transformation owns all properties
• conflictMode: upsert — most common; creates or updates records
• Authentication uses Toolkit variables from config.<env>.yaml
• query: can inline a short SQL string; for longer queries use a paired .sql file

---

Destination Types

RAW Staging Area

destination:
  type: raw
  rawDatabase: my_database
  rawTable: my_table

Data Modeling — Nodes (single view)

destination:
  type: nodes
  view:
    space: '{{schema_space}}'
    externalId: Pump
    version: '{{model_version}}'
  instanceSpace: '{{instance_space}}'

Data Modeling — Edges

destination:
  type: edges
  view:
    space: '{{schema_space}}'
    externalId: PumpToValve
    version: '{{model_version}}'
  instanceSpace: '{{instance_space}}'
  edgeType:
    space: '{{schema_space}}'
    externalId: pump_to_valve

Data Modeling — Instances (full data model / specific type)

destination:
  type: instances
  dataModel:
    space: APM_SourceData
    externalId: APM_SourceData
    version: "1"
    destinationType: APM_Activity
  instanceSpace: '{{instance_space}}'

---

Schedule YAML (.schedule.yaml)

externalId: tr_my_transform
interval: '{{scheduleHourly}}'   # cron expression or Toolkit variable
isPaused: false

Common cron examples: '0 * * * *' (hourly), '0 0 * * *' (daily midnight), '*/15 * * * *' (every 15 min).

Scheduling guidance

• Stagger schedules across transformations. Don't run every transformation on 0 * * * * — they will all hit the cluster at the same instant and contend for resources. Spread start minutes (e.g., 7 * * * *, 19 * * * *, 34 * * * *) so the load smooths out.
• Match the interval to source data velocity. Fast-moving sources (alarms, sensor values) deserve short intervals; static reference data (sites, equipment classes) does not.
• is_new() makes frequent schedules cheap. A correctly incrementalized transformation with no new rows finishes in seconds, so a 5-minute interval on a quiet table is not wasteful.
• Prefer Workflows over standalone schedules when transformations have ordering dependencies (see Best Practices). For Workflow design, scheduling, and task-type guidance, apply the cognite-workflow skill from this plugin.

---

SQL — Reading Data Sources

From RAW (most common)

SELECT *
FROM `my_database`.`my_table`

Use backticks for names with hyphens or spaces.

Schema inference caveat: Transformations infer the RAW table schema from a subset of rows. If your data is heterogeneous or sparsely populated, inferred types may be wrong or columns may be missing entirely. Only use cdf_raw() in this case — prefer standard table syntax in all other situations. With cdf_raw(), parse the JSON columns field manually:

-- cdf_raw() returns: key (STRING), lastUpdatedTime (TIMESTAMP), columns (JSON STRING)
SELECT
  get_json_object(columns, '$.externalId')            AS externalId,
  get_json_object(columns, '$.name')                  AS name,
  cast(get_json_object(columns, '$.value') AS DOUBLE) AS value,
  to_timestamp(
    cast(get_json_object(columns, '$.ts') AS LONG) / 1000
  )                                                   AS timestamp
FROM cdf_raw('my_database', 'my_table')

From DMS Nodes/Edges

-- All nodes from a view
SELECT * FROM cdf_nodes('space', 'ViewExternalId', 'version')

-- All edges
SELECT * FROM cdf_edges('space', 'ViewExternalId', 'version')

---

SQL — Style Guide

Consistent style makes transformations easier to read and review.

• Uppercase all SQL keywords: SELECT, FROM, WHERE, JOIN, ON, AND, OR, WITH, AS, CASE, WHEN, THEN, ELSE, END, UNION ALL, NULL, IS NULL, IS NOT NULL, CAST, DISTINCT
• Lowercase all CDF built-in functions: node_reference(), is_new(), dataset_id(), to_metadata_except(), etc.
• Lowercase column and table references
• Align AS aliases vertically for multi-column SELECTs
• One column per line in SELECT
• Leading commas — place the comma at the start of each column line (not the end). This makes it trivial to comment out individual columns when debugging without breaking the trailing-comma syntax.
• CTEs over subqueries — use WITH clauses to name intermediate steps rather than nesting subqueries
• Explicit column list — never use SELECT * in the final destination query; always name each output column
• Backticks for awkward names — RAW column or table names containing spaces, hyphens, or reserved words must be wrapped in backticks (`Field Name`), never in single or double quotes
• Indentation and width — keep queries readable within a typical editor width; wrap long expressions and align continuations
• Inline comments — annotate complex JOIN conditions, CASE logic, or any non-obvious filter with -- comments so reviewers can follow the intent

Example:

WITH source AS (
  SELECT *
  FROM `{{raw_db}}`.`{{raw_table}}`
  WHERE is_new('source_version', lastUpdatedTime)
)
SELECT
    cast(externalId AS STRING)                        AS externalId
  , cast(name AS STRING)                              AS name
  , coalesce(cast(description AS STRING), '')         AS description
  , node_reference('{{instance_space}}', parentId)    AS parent
FROM source
WHERE externalId IS NOT NULL

---

SQL — Syntax Reference

Type Casting

All columns written to DMS must match the target container property type. Explicit casting avoids type inference surprises:

cast(col AS STRING)
cast(col AS DOUBLE)
cast(col AS LONG)
cast(col AS BOOLEAN)
cast(col AS TIMESTAMP)

Handling Nulls

coalesce(col, 'default')           -- first non-null value
nullif(col, '')                    -- return null if value equals ''
if(col IS NULL, 'fallback', col)   -- inline conditional

JSON Columns (common in RAW)

-- Extract a single field from a JSON string column
get_json_object(col, '$.fieldName')
get_json_object(col, '$.nested.field')

-- Parse a JSON string into a struct (provide schema)
from_json(col, 'struct<name:string, value:double>')

-- Convert a struct/map back to JSON string
to_json(struct_col)

Timestamps

RAW data often stores timestamps as epoch milliseconds (LONG). Convert before writing to a TIMESTAMP property:

-- Epoch milliseconds to timestamp
to_timestamp(col / 1000)

-- Epoch seconds to timestamp
from_unixtime(col)

-- String to timestamp
to_timestamp(col, 'yyyy-MM-dd HH:mm:ss')

-- Timestamp to epoch milliseconds (e.g. for datapoints)
unix_timestamp(col) * 1000

String Operations

concat(a, '-', b)
trim(col)
lower(col)
regexp_replace(col, '[^a-zA-Z0-9]', '_')   -- sanitise for externalId

CTEs (WITH clauses)

Use CTEs to break complex queries into readable steps:

WITH source AS (
  SELECT *
  FROM `{{raw_db}}`.`{{raw_table}}`
  WHERE is_new('source_version', lastUpdatedTime)
),
enriched AS (
  SELECT
    s.externalId,
    s.name,
    coalesce(s.description, '') AS description
  FROM source s
)
SELECT * FROM enriched

List Properties (array)

DMS list-type properties expect an array:

array(val1, val2)                          -- literal array
split(col, ',')                            -- split string into array

Direct Relations via struct