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
13 changes: 9 additions & 4 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -4,15 +4,20 @@

### Added

- **Streaming (delta) read/write methods on `Client` for BYOC streaming transforms.**
- **`StreamingClient` for BYOC streaming (delta) transforms.**

New methods let an entry point process a Data Lake Object's Change Data Feed continuously instead of reading a bounded snapshot:
A dedicated `StreamingClient` (alongside the batch `Client`) lets an entry point process a Data Lake Object's Change Data Feed continuously instead of reading a bounded snapshot.

- `read_dlo_deltas(name)` / `read_dmo_deltas(name)` – return a streaming DataFrame over the object's change feed.
- `read_dlo_deltas()` / `read_dmo_deltas()` – return a streaming DataFrame over the object's change feed.
- `write_dlo_deltas(name, dataframe)` – start a streaming query that writes each micro-batch to the target DLO and return the `StreamingQuery` handle.

The shared functions (`find_file_path`, `llm_gateway_generate_text`, `einstein_predict`) are available on both `Client` and `StreamingClient`.

```python
deltas = client.read_dlo_deltas("Input__dll")
from datacustomcode import StreamingClient

client = StreamingClient()
deltas = client.read_dlo_deltas()
transformed = deltas.withColumn("description__c", upper(col("description__c")))
query = client.write_dlo_deltas("Output__dll", transformed)
query.awaitTermination()
Expand Down
21 changes: 12 additions & 9 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -145,20 +145,22 @@ Your Python dependencies can be packaged as .py files, .zip archives (containing

## API

Your entry point script will define logic using the `Client` object which wraps data access layers.
Your entry point script will define logic using the `Client` object (for batch transforms) or the `StreamingClient` object (for streaming delta transforms), which wrap the data access layers. Both are singletons; a single transform should use one or the other, not both.

You should only need the following methods:
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

For streaming (delta) transforms, the streaming counterparts are:
* `read_dlo_deltas(name)` – Read the streaming change feed (deltas) of a Data Lake Object as a streaming DataFrame
* `read_dmo_deltas(name)` – Read the streaming change feed (deltas) of a Data Model Object as a streaming 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.
* `read_dmo_deltas()` – Read the streaming change feed (deltas) of a Data Model Object as a streaming DataFrame.
* `write_dlo_deltas(name, spark_dataframe)` – Write a streaming DataFrame of deltas to a Data Lake Object; returns the started `StreamingQuery`

`find_file_path`, `llm_gateway_generate_text`, and `einstein_predict` are available on both clients.

For example:
```python
from datacustomcode import Client
Expand All @@ -176,17 +178,18 @@ client.write_to_dlo('output_DLO')

### Streaming (delta) transforms

Streaming BYOC transforms process a Data Lake Object's Change Data Feed continuously instead of reading a bounded snapshot. Use the `*_deltas` methods in place of the batch read/write methods:
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:

```python
from pyspark.sql.functions import col, upper

from datacustomcode import Client
from datacustomcode import StreamingClient

client = Client()
client = StreamingClient()

# read_dlo_deltas returns a *streaming* DataFrame over the change feed.
deltas = client.read_dlo_deltas("Input__dll")
# The runtime resolves the single streaming source, so no name is passed.
deltas = client.read_dlo_deltas()

# Ordinary PySpark transform.
transformed = deltas.withColumn("description__c", upper(col("description__c")))
Expand Down
5 changes: 5 additions & 0 deletions src/datacustomcode/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@
"QueryAPIDataCloudReader",
"SparkEinsteinPredictions",
"SparkLLMGateway",
"StreamingClient",
"einstein_predict_col",
"llm_gateway_generate_text_col",
]
Expand All @@ -34,6 +35,10 @@ def __getattr__(name: str):
from datacustomcode.client import Client

return Client
elif name == "StreamingClient":
from datacustomcode.client import StreamingClient

return StreamingClient
elif name == "AuthType":
from datacustomcode.credentials import AuthType

Expand Down
Loading
Loading