Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 23 additions & 2 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -151,8 +151,8 @@ For a batch transform, use `Client`. You should only need the following methods:
* `find_file_path(file_name)` – Resolve a bundled file (placed under `payload/files/`) to a `pathlib.Path` that exists. Works the same locally and inside Data Cloud — see [Bundled file resolution](#bundled-file-resolution) below for the full lookup order. Raises `FileNotFoundError` if the file isn't found.
* `read_dlo(name)` – Read from a Data Lake Object by name
* `read_dmo(name)` – Read from a Data Model Object by name
* `write_to_dlo(name, spark_dataframe, write_mode)` – Write to a Data Model Object by name with a Spark dataframe
* `write_to_dmo(name, spark_dataframe, write_mode)` – Write to a Data Lake Object by name with a Spark dataframe
* `write_to_dlo(name, spark_dataframe, write_mode)` – Write to a Data Lake Object by name with a Spark dataframe
* `write_to_dmo(name, spark_dataframe, write_mode)` – Write to a Data Model Object by name with a Spark dataframe

For a streaming (delta) transform, use `StreamingClient`, which exposes the streaming counterparts:
* `read_dlo_deltas()` – Read the streaming change feed (deltas) of a Data Lake Object as a streaming DataFrame.
Expand All @@ -176,6 +176,27 @@ client.write_to_dlo('output_DLO')
> [!WARNING]
> Currently we only support reading from DMOs and writing to DMOs or reading from DLOs and writing to DLOs, but they cannot mix.

### Writing Currency fields

A DLO or DMO with a Currency field has a system field, `cdp_sys_record_currency__c`, that holds each record's ISO currency code. Data Explorer UI uses it to display currency values, and shows a blank value if it's null (Query Editor still shows it).

Code Extensions don't set `cdp_sys_record_currency__c` for you, since only your script knows what currency its amounts are in. When writing to an object with Currency fields, set it yourself:

```python
from pyspark.sql.functions import lit

# Either source has Currency fields: keep its cdp_sys_record_currency__c in your output
df = source.select("id__c", "amount__c", "cdp_sys_record_currency__c")

# OR amounts are in a known currency
df = df.withColumn("cdp_sys_record_currency__c", lit("EUR"))
```

- Use an ISO code that exists in your org's configured currency rates.
- When aggregating, group by the currency code so amounts in different currencies aren't summed together.

See [`docs/examples/script/currency_totals`](docs/examples/script/currency_totals) for a complete example.

### Streaming (delta) transforms

Streaming BYOC transforms process a Data Lake Object's Change Data Feed continuously instead of reading a bounded snapshot. Use a `StreamingClient` and its `*_deltas` methods in place of the batch `Client` read/write methods:
Expand Down
33 changes: 33 additions & 0 deletions docs/examples/script/currency_totals/README.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,33 @@
# Example: Totals that keep their currency code

A Code Extension for Batch Transform that sums `Payment__dll` amounts per customer and writes them to `Customer_Payment_Totals__dll`, a DLO with a Currency field, setting the `cdp_sys_record_currency__c` system field so Data Explorer UI can display the totals.

## What it demonstrates

- **Carrying `cdp_sys_record_currency__c` through** from the source rows to the output. Code Extensions don't auto-populate `cdp_sys_record_currency__c`. And if it's null, Data Explorer shows the Currency value as blank. See [Writing Currency fields](../../../../README.md#writing-currency-fields).
- **Grouping by currency** so EUR and USD amounts are never summed together. Customer `C1` gets one EUR row and one USD row.
- If all amounts are in one known currency, one can also use `lit("EUR")` instead of copying a column.

## Prerequisites

Input DLO `Payment__dll` in the target dataspace:

| Column | Type | Notes |
|------------------------------|----------|------------------------------------------------------------------------|
| `id__c` | text | Primary key. |
| `customer_id__c` | text | |
| `amount__c` | Currency | |
| `cdp_sys_record_currency__c` | text | System field, added with the Currency field. From CSV `currency_code`. |

Data should be loaded via `sample_data/payment.csv`. See [loading sample data](../README.md#loading-sample-data).

Output DLO `Customer_Payment_Totals__dll` in the same dataspace:

| Column | Type | Notes |
|------------------------------|----------|--------------------------------------------------|
| `id__c` | text | Primary key (`<customer_id>_<currency>`). |
| `customer_id__c` | text | |
| `total_amount__c` | Currency | |
| `cdp_sys_record_currency__c` | text | System field, added with the Currency field. |

The ISO codes in `sample_data/payment.csv` must exist in your org's currency rates.
13 changes: 13 additions & 0 deletions docs/examples/script/currency_totals/payload/config.json
Original file line number Diff line number Diff line change
@@ -0,0 +1,13 @@
{
"sdkVersion": "6.1.0",
"entryPoint": "entrypoint.py",
"dataspace": "default",
"permissions": {
"read": {
"dlo": ["Payment__dll"]
},
"write": {
"dlo": ["Customer_Payment_Totals__dll"]
}
}
}
35 changes: 35 additions & 0 deletions docs/examples/script/currency_totals/payload/entrypoint.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,35 @@
from pyspark.sql.functions import (
col,
concat_ws,
sum as _sum,
)

from datacustomcode.client import Client
from datacustomcode.io.writer.base import WriteMode


def main():
client = Client()

payments = client.read_dlo("Payment__dll").select(
"customer_id__c", "amount__c", "cdp_sys_record_currency__c"
)

totals = (
payments.groupBy("customer_id__c", "cdp_sys_record_currency__c")
.agg(_sum("amount__c").alias("total_amount__c"))
.select(
concat_ws(
"_", col("customer_id__c"), col("cdp_sys_record_currency__c")
).alias("id__c"),
col("customer_id__c"),
col("total_amount__c"),
col("cdp_sys_record_currency__c"),
)
)

client.write_to_dlo("Customer_Payment_Totals__dll", totals, WriteMode.OVERWRITE)


if __name__ == "__main__":
main()
1 change: 1 addition & 0 deletions docs/examples/script/currency_totals/requirements.txt
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
# Packages required for the custom code
7 changes: 7 additions & 0 deletions docs/examples/script/currency_totals/sample_data/payment.csv
Original file line number Diff line number Diff line change
@@ -0,0 +1,7 @@
id,customer_id,amount,currency_code
P1,C1,100.00,EUR
P2,C1,250.50,EUR
P3,C1,40.00,USD
P4,C2,1200.00,USD
P5,C2,300.00,USD
P6,C3,75.25,GBP
Loading