cognite-transformation

> **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. ```sql -- 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`) ```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 ```yaml destination: type: raw rawDatabase: my_database rawTable: my_table ``` ## Data Modeling — Nodes (single view) ```yaml destination: type: nodes view: space: '{{schema_space}}' externalId: Pump version: '{{model_version}}' instanceSpace: '{{instance_space}}' ``` ## Data Modeling — Edges ```yaml 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) ```yaml destination: type: instances dataModel: space: APM_SourceData externalId: APM_SourceData version: "1" destinationType: APM_Activity instanceSpace: '{{instance_space}}' ``` --- # Schedule YAML (`.schedule.yaml`) ```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`](../cdf-workflow/SKILL.md) skill from this plugin. --- # SQL — Reading Data Sources ## From RAW (most common) ```sql 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: ```sql -- 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 ```sql -- 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:** ```sql 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: ```sql cast(col AS STRING) cast(col AS DOUBLE) cast(col AS LONG) cast(col AS BOOLEAN) cast(col AS TIMESTAMP) ``` ## Handling Nulls ```sql 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) ```sql -- 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: ```sql -- 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 ```sql 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: ```sql 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: ```sql array(val1, val2) -- literal array split(col, ',') -- split string into array ``` ## Direct Relations via struct