diff --git a/README.md b/README.md index bf9e929..b258532 100644 --- a/README.md +++ b/README.md @@ -4,21 +4,49 @@ Python SDK for integrating Databricks with The Trade Desk Data API. Supports First Party Data, Third Party Data, Offline Conversion Data, and Deletion/Opt-Out workflows. +**Key features:** + - **Ad hoc mode** — push a DataFrame directly and receive per-row results inline - **Batch mode** — run incremental pipelines backed by Delta tables with processing checkpoints -- Built-in schema validation and per-row error tracking +- **Schema validation and error tracking** — required columns are checked before submission, and every row comes back with its own success or error status ## Table of Contents +- [Example Notebooks](#example-notebooks) - [SDK Installation](#sdk-installation) -- [Initial Setup](#initial-setup) -- [SDK Example Usage](#sdk-example-usage) -- [Authentication](#authentication) -- [Supported Data API Endpoints](#supported-data-api-endpoints) -- [UID2 Support](#uid2-support) +- [Quickstart](#quickstart) + - [1. Create a Client](#1-create-a-client) + - [Authentication](#authentication) + - [2. Create a Context](#2-create-a-context) + - [3. Inspect the Schema and Prepare Your Input DataFrame](#3-inspect-the-schema-and-prepare-your-input-dataframe) + - [4. Send the Data](#4-send-the-data) + - [4a. Ad Hoc — `push_data`](#4a-ad-hoc--push_data) + - [4b. Batch Processing — `batch_process`](#4b-batch-processing--batch_process) +- [Supported Data API Endpoints](#supported-data-api-endpoints) — by use case: + - [First-Party Data (1P)](#first-party-data--dataadvertiser) — `/data/advertiser` + - [Third-Party Data (3P)](#third-party-data--datathirdparty) — `/data/thirdparty` + - [Offline Conversion (CAPI)](#offline-conversion--providerapiofflineconversion) — `/providerapi/offlineconversion` + - [Deletion / Opt-Out — Advertiser](#deletion--opt-out--advertiser--datadeletion-optoutadvertiser) — `/data/deletion-optout/advertiser` + - [Deletion / Opt-Out — Third Party](#deletion--opt-out--third-party--datadeletion-optoutthirdparty) — `/data/deletion-optout/thirdparty` + - [Deletion / Opt-Out — Merchant](#deletion--opt-out--merchant--datadeletion-optoutmerchant) — `/data/deletion-optout/merchant` - [Error Handling](#error-handling) -- [Server Selection](#server-selection) -- [Custom HTTP Client](#custom-http-client) +- [Optional Configuration](#optional-configuration) + - [UID2 Support](#uid2-support) + - [Server Selection](#server-selection) + - [Custom HTTP Client](#custom-http-client) + +## Example Notebooks + +The following table maps each use case supported by the SDK to the Trade Desk endpoint its data is sent to, and to a quickstart example notebook. + +The example notebooks are for users who want to dive straight in and try the SDK — each one is runnable and covers the whole flow, from credentials through to reading per-row results. The sections after this break the same integration down step by step. + +| Use case | Destination Endpoint to Which Data Is Sent | Example Notebook | +|---|---|---| +| First-party data (1P) | `POST /data/advertiser` | [First Party Data (1PD) Example Notebook.ipynb](https://github.com/thetradedesk/ttd-databricks-python/blob/add-easy-start-examples-per-usecase-to-sdk-docs/example_notebook/First%20Party%20Data%20%281PD%29%20Example%20Notebook.ipynb) | +| Third-party data (3P) | `POST /data/thirdparty` | [Third Party Data (3PD) Example Notebook.ipynb](https://github.com/thetradedesk/ttd-databricks-python/blob/add-easy-start-examples-per-usecase-to-sdk-docs/example_notebook/Third%20Party%20Data%20%283PD%29%20Example%20Notebook.ipynb) | +| Offline conversion (CAPI) | `POST /providerapi/offlineconversion` | [Offline Conversion Data (CAPI) Example Notebook.ipynb](https://github.com/thetradedesk/ttd-databricks-python/blob/add-easy-start-examples-per-usecase-to-sdk-docs/example_notebook/Offline%20Conversion%20Data%20%28CAPI%29%20Example%20Notebook.ipynb) | +| Deletion and opt-out | `POST /data/deletion-optout/*` | [Deletion and Opt-Out (DSR) Example Notebook.ipynb](https://github.com/thetradedesk/ttd-databricks-python/blob/add-easy-start-examples-per-usecase-to-sdk-docs/example_notebook/Deletion%20and%20Opt-Out%20%28DSR%29%20Example%20Notebook.ipynb) | --- @@ -32,20 +60,47 @@ Requires Python 3.10 or higher. Intended to run inside a Databricks environment --- -## Initial Setup +## Quickstart + +The following steps break down the process of integrating with the ttd-databricks SDK, using first-party data as the worked example. Every other use case follows the same steps with a different context and different input columns — see [Supported Data API Endpoints](#supported-data-api-endpoints) for the use cases supported and the example notebook for each. ### 1. Create a Client -The client is the entry point for all SDK operations. Create it once and reuse it across calls. +The client is the entry point for all SDK operations. Create it once and reuse it across calls. There are two ways to create it — pick one. + +**i) Factory — `from_params` builds the `DataClient` for you:** + +```python +from ttd_databricks_python.ttd_databricks import TtdDatabricksClient + +client = TtdDatabricksClient.from_params( + api_token="", + # spark=spark, # optional; auto-detected from the Databricks runtime + # server_url="https://...", # optional; see Server Selection + # retry_config=RetryConfig(...), # optional; 429/5xx are retried by default, None disables + # timeout_ms=10000, # optional; per-request timeout in milliseconds +) +``` + +**ii) Dependency injection — you build the `DataClient` and pass it in:** ```python +from ttd_data import DataClient from ttd_databricks_python.ttd_databricks import TtdDatabricksClient -# SparkSession is auto-detected from the Databricks runtime if not provided. -client = TtdDatabricksClient.from_params(api_token="") +client = TtdDatabricksClient( + data_api_client=DataClient(), + api_token="", +) ``` -See [Authentication](#authentication) for alternative client creation options. +The rest of this README uses (i). To configure the underlying HTTP transport, or to inject a mock in tests, use (ii) — see [Custom HTTP Client](#custom-http-client). + +#### Authentication + +All underlying API calls made within the SDK authenticate with a TTD API token, passed as `api_token` at client creation as shown above and sent as the `TTD-Auth` header. The SDK does not support `TtdSignature` based authentication. + +See [OpenTTD](https://open.thetradedesk.com/advertiser/docsApp/Foundations/resources/doc/PlatformAuthentication) for instructions on how to create your API token. --- @@ -56,7 +111,7 @@ A context specifies which TTD endpoint to target and carries the identifiers (ad ```python from ttd_databricks_python.ttd_databricks import AdvertiserContext -# Each endpoint has its own context class. See Available Resources and Operations +# Each endpoint has its own context class. See Supported Data API Endpoints # for the full list. context = AdvertiserContext( advertiser_id="", @@ -66,9 +121,113 @@ context = AdvertiserContext( --- -### 3. Set Up Delta Tables +### 3. Inspect the Schema and Prepare Your Input DataFrame -If you plan to use batch processing, use the following helpers to set up the input, output, and metadata Delta tables. These can be created once and reused for all future executions. +Each endpoint has its own input schema. Retrieve it, and the subset of columns that are mandatory, straight from the SDK: + +```python +from ttd_databricks_python.ttd_databricks import TTDEndpoint, get_ttd_input_schema +from ttd_databricks_python.ttd_databricks.schemas import get_required_column_names + +input_schema = get_ttd_input_schema(TTDEndpoint.ADVERTISER) + +for field in input_schema.fields: + print(f"{field.name}: {field.dataType.simpleString()} (nullable={field.nullable})") + +required_cols = get_required_column_names(TTDEndpoint.ADVERTISER) +# e.g. ["id_type", "id_value", "segment_name"] +``` + +Nullable columns may be omitted from your DataFrame — they are filled with null automatically. + +Now build the DataFrame. Always pass `schema=`. Without it Spark infers the types, and nested columns (such as offline conversion's `user_ids`) come out as `MapType` instead of the `array` the API requires. + +```python +from ttd_databricks_python.ttd_databricks import TTDEndpoint, get_ttd_input_schema + +input_schema = get_ttd_input_schema(TTDEndpoint.ADVERTISER) + +rows = [ + {"id_type": "TDID", "id_value": "123e4567-e89b-12d3-a456-426652340000", + "segment_name": "my_first_segment", "ttl_in_minutes": 43200}, + {"id_type": "DAID", "id_value": "a9342d1f-69f1-4bf8-bc2b-1f20eb451f21", + "segment_name": "my_first_segment", "ttl_in_minutes": 43200}, + {"id_type": "UID2", "id_value": "48MjlfIUZpOKNAm9nod7/jCLAXUYsnE1tpVHQSDS0uo=", + "segment_name": "my_first_segment", "ttl_in_minutes": 43200}, +] + +# spark is the SparkSession available in the Databricks notebook runtime. +input_df = spark.createDataFrame(rows, schema=input_schema) +``` + +Optionally, pre-validate the DataFrame to catch missing columns before you send anything: + +```python +from ttd_databricks_python.ttd_databricks import TTDEndpoint +from ttd_databricks_python.ttd_databricks.schemas import validate_ttd_schema + +# Raises TTDSchemaValidationError if any required columns are missing. +validate_ttd_schema(df=input_df, endpoint=TTDEndpoint.ADVERTISER) +``` + +> **Tip:** Start with a handful of rows. Confirming the end-to-end flow on a small sample is much easier to troubleshoot than a full load. + +--- + +### 4. Send the Data + +There are two ways to send your data. Pick one — you do not need both. + +| | Ad hoc (`push_data`) | Batch processing (`batch_process`) | +|---|---|---| +| **State management** | None. Every call sends every row you give it. | Provided. A metadata table records progress, so each run sends only rows added since the last one. | +| **Input** | A DataFrame you build in the notebook | A Delta input table | +| **Output** | Returned inline as a DataFrame | Written to a Delta output table | +| **Best for** | One-off loads and first tests | Recurring pipelines | + +Neither raises on API or row-level failures — every outcome is reported inline on the row. + +#### 4a. Ad Hoc — `push_data` + +`push_data` submits the DataFrame in batches and returns your input columns enriched with per-row status. + +```python +result_df = client.push_data( + df=input_df, + context=context, + batch_size=1600, # number of rows per API request + # data_load_trace_id="", # optional; sent as DataLoadTraceId, for debugging +) +``` + +`push_data` adds these columns to your input: + +| Column | Meaning | +|---|---| +| `success` | `True` if the row was accepted | +| `error_code` | Failure category, `null` on success | +| `error_message` | Human-readable reason, `null` on success | +| `processed_timestamp` | When the row was submitted | +| `uid2_resolutions` | Raw identifier to UID2 mapping, empty unless `uid2_config` was set | + +```python +from pyspark.sql.functions import col + +total = result_df.count() +succeeded = result_df.filter(col("success")).count() + +print(f"Total: {total} | Succeeded: {succeeded} | Failed: {total - succeeded}") + +failed_df = result_df.filter(~col("success")) +if failed_df.count(): + failed_df.select("error_code", "error_message").show(truncate=False) +``` + +#### 4b. Batch Processing — `batch_process` + +Use this for incremental, distributed processing backed by Delta tables. Only records added since the last run are sent. + +**One time steps:** Create the input, output, and metadata Delta tables. These are created once and reused by every future run — the metadata table is what tracks how far the last run got, so do not drop or recreate it between runs. The `setup_*` helpers return the existing table if it is already there, so they are safe to re-run. ```python from ttd_databricks_python.ttd_databricks import TTDEndpoint @@ -79,7 +238,7 @@ from ttd_databricks_python.ttd_databricks import TTDEndpoint input_table = client.setup_input_table(endpoint=TTDEndpoint.ADVERTISER) # Output table: mirrors the input schema plus status columns -# (success, error_code, error_message, processed_timestamp). +# (success, error_code, error_message, processed_timestamp, uid2_resolutions). # Default table name: ttd_{endpoint}_output (e.g. "ttd_advertiser_output"). output_table = client.setup_output_table(endpoint=TTDEndpoint.ADVERTISER) @@ -98,53 +257,36 @@ input_table = client.setup_input_table( ) ``` ---- - -## SDK Example Usage - -The SDK supports two processing modes. +**Every run:** -### Ad Hoc Mode (`push_data`) +**1. Append new rows to the input table.** `setup_input_table` creates it empty — the SDK does +not populate it. In production this is your upstream pipeline's job; in a notebook it is a +DataFrame appended to the table. -Use this to process a DataFrame directly and receive results inline. +Set `updated_at` on every row you append. That column is what `process_new_records_only=True` +filters on, so rows without it are never picked up incrementally. ```python -from ttd_databricks_python.ttd_databricks import ( - TtdDatabricksClient, - AdvertiserContext, -) +from pyspark.sql import functions as F +from ttd_databricks_python.ttd_databricks import TTDEndpoint, get_ttd_input_schema -# Create the client using your TTD auth token. -# SparkSession is auto-detected from the Databricks runtime if not provided. -client = TtdDatabricksClient.from_params(api_token="") +input_schema = get_ttd_input_schema(TTDEndpoint.ADVERTISER) -# Create a context for the target endpoint. -# The context identifies which advertiser/provider to push data to -# and is passed to every API call. -context = AdvertiserContext( - advertiser_id="", - data_provider_id="", # optional -) +rows = [ + {"id_type": "TDID", "id_value": "123e4567-e89b-12d3-a456-426652340000", + "segment_name": "my_first_segment", "ttl_in_minutes": 43200}, +] -# Push the DataFrame to the TTD Data API in batches. -# Returns the input DataFrame enriched with status columns. -result_df = client.push_data( - df=input_df, - context=context, - batch_size=1600, # number of rows per API request +( + spark.createDataFrame(rows, schema=input_schema) + .withColumn("updated_at", F.current_timestamp()) + .write.format("delta").mode("append").saveAsTable(input_table) ) -# result_df contains all input columns plus: -# success, error_code, error_message, processed_timestamp ``` -### Batch Processing Mode (`batch_process`) - -Use this for incremental, distributed processing backed by Delta tables. Supports incremental filtering to process only records added since the last run. +**2. Call `batch_process`.** Re-running it picks up only rows appended since the last run. ```python -# Tables set up during Initial Setup (see above). -# input_table, output_table, metadata_table already created. - # Run the batch pipeline. With process_new_records_only=True, only rows # added since the last successful run (tracked via metadata_table) are sent. client.batch_process( @@ -153,7 +295,9 @@ client.batch_process( output_table=output_table, metadata_table=metadata_table, process_new_records_only=True, # incremental; set False to reprocess all rows - batch_size=1600, # rows per API request + batch_size=1600, # rows per API request + parallelism=8, # parallel partitions for API calls; default 8 + # data_load_trace_id="", # optional; sent as DataLoadTraceId, for debugging ) ``` @@ -174,53 +318,11 @@ client.batch_process( --- -## Authentication - -All API calls require a TTD auth token passed at client creation time. - -> **Note:** The SDK does not support `TtdSignature` based authentication. Refer to [OpenTTD](https://open.thetradedesk.com/advertiser/docsApp/Foundations/resources/doc/PlatformAuthentication) for instructions on how to create your `TTD-Auth` API token (select `Data API` as the `Application`). - -### Factory Method (recommended for notebooks) - -```python -# spark is the SparkSession available in the Databricks notebook runtime. -client = TtdDatabricksClient.from_params( - api_token="", # your TTD platform API token - spark=spark, # optional; auto-detected from Databricks context - # server_url="https://..." # optional; see Server Selection - # retry_config=RetryConfig(...) # optional; transient errors (429/5xx) are retried by - # default, pass None to disable. See Custom HTTP Client - # timeout_ms=10000 # optional; per-request timeout in milliseconds -) -``` - -### Dependency Injection (recommended for testing) - -Provide your own [`DataClient`](https://github.com/thetradedesk/ttd-data-python/blob/main/src/ttd_data/sdk.py) instance to control the underlying HTTP transport directly. -Use this when you need to configure options not exposed by `from_params()`, or to inject a mock in tests. - -```python -from ttd_data import DataClient -from ttd_databricks_python.ttd_databricks import TtdDatabricksClient - -# Configure DataClient with custom HTTP settings. -data_client = DataClient( - server_url="https://custom-server.example.com", # override default server URL - timeout_ms=10000, # request timeout in milliseconds -) - -client = TtdDatabricksClient( - data_api_client=data_client, - api_token="", - spark=spark, # optional; spark variable available from the Databricks notebook runtime -) -``` - ---- - ## Supported Data API Endpoints -Each Data API endpoint is represented by a context dataclass that configures the API call. The following table lists the supported Data API endpoints, their corresponding SDK contexts, and OpenTTD documentation. +Each Data API endpoint is represented by a context dataclass. You never pass an endpoint yourself: choosing a context selects the endpoint, its request shape, and its default server. The endpoints are listed here for reference, to connect each SDK context to the API documentation that describes it. + +Each section below gives the context, its mandatory columns, a sample input DataFrame, and the example notebook for that use case. The three deletion/opt-out endpoints share one input shape, so the sample appears once, under Advertiser. To list an endpoint's columns programmatically or validate a DataFrame before sending, see [Quickstart step 3](#3-inspect-the-schema-and-prepare-your-input-dataframe). | Data API Endpoint | Context | OpenTTD API Documentation | |---|---|---| @@ -244,7 +346,11 @@ context = AdvertiserContext( ) ``` -The input schema for `TTDEndpoint.ADVERTISER` is defined in [advertiser.py](https://github.com/thetradedesk/ttd-databricks-python/blob/main/ttd_databricks_python/ttd_databricks/schemas/advertiser.py). See the [Inspecting Schemas](#inspecting-schemas) section for helper functions to view columns and validate your DataFrame. +**Mandatory columns:** `id_type`, `id_value`, `segment_name`. The Quickstart uses this endpoint — see [step 3](#3-inspect-the-schema-and-prepare-your-input-dataframe) for a sample input DataFrame. + +**Example notebook:** [First Party Data (1PD) Example Notebook.ipynb](https://github.com/thetradedesk/ttd-databricks-python/blob/add-easy-start-examples-per-usecase-to-sdk-docs/example_notebook/First%20Party%20Data%20%281PD%29%20Example%20Notebook.ipynb). + +**Schema:** [advertiser.py](https://github.com/thetradedesk/ttd-databricks-python/blob/add-easy-start-examples-per-usecase-to-sdk-docs/ttd_databricks_python/ttd_databricks/schemas/advertiser.py). --- @@ -261,7 +367,26 @@ context = ThirdPartyContext( ) ``` -The input schema for `TTDEndpoint.THIRD_PARTY` is defined in [third_party.py](https://github.com/thetradedesk/ttd-databricks-python/blob/main/ttd_databricks_python/ttd_databricks/schemas/third_party.py). See the [Inspecting Schemas](#inspecting-schemas) section for helper functions to view columns and validate your DataFrame. +**Mandatory columns:** `id_type`, `id_value`, `segment_name`. `segment_name` is your third-party segment identifier. + +```python +input_schema = get_ttd_input_schema(TTDEndpoint.THIRD_PARTY) + +rows = [ + {"id_type": "TDID", "id_value": "123e4567-e89b-12d3-a456-426652340000", + "segment_name": "1210", "ttl_in_minutes": 43200}, + {"id_type": "ID5", "id_value": "ID5-c62drGF0EC6wsCZVFDbTbZwi33eB0uZTIC8FxJpzsQ", + "segment_name": "1800", "ttl_in_minutes": 43200}, + {"id_type": "FirstId", "id_value": "8934d279bba4c7d652a02f624dc334e3", + "segment_name": "1810", "ttl_in_minutes": 43200}, +] + +input_df = spark.createDataFrame(rows, schema=input_schema) +``` + +**Example notebook:** [Third Party Data (3PD) Example Notebook.ipynb](https://github.com/thetradedesk/ttd-databricks-python/blob/add-easy-start-examples-per-usecase-to-sdk-docs/example_notebook/Third%20Party%20Data%20%283PD%29%20Example%20Notebook.ipynb). + +**Schema:** [third_party.py](https://github.com/thetradedesk/ttd-databricks-python/blob/add-easy-start-examples-per-usecase-to-sdk-docs/ttd_databricks_python/ttd_databricks/schemas/third_party.py). --- @@ -276,7 +401,31 @@ context = OfflineConversionContext( ) ``` -The input schema for `TTDEndpoint.OFFLINE_CONVERSION` is defined in [offline_conversion.py](https://github.com/thetradedesk/ttd-databricks-python/blob/main/ttd_databricks_python/ttd_databricks/schemas/offline_conversion.py). See the [Inspecting Schemas](#inspecting-schemas) section for helper functions to view columns and validate your DataFrame. +**Mandatory columns:** `tracking_tag_id`, `timestamp_utc`. This endpoint takes a different shape from the audience endpoints — one row per conversion **event**, with identities nested in `user_ids` rather than flat `id_type`/`id_value` columns. `user_ids` is required unless `impression_id` is provided. + +```python +from datetime import datetime, timezone + +input_schema = get_ttd_input_schema(TTDEndpoint.OFFLINE_CONVERSION) + +rows = [ + {"tracking_tag_id": "", + "timestamp_utc": datetime(2026, 1, 15, 10, 11, 30, tzinfo=timezone.utc), + "user_ids": [{"type": "TDID", "id": "123e4567-e89b-12d3-a456-426652340000"}]}, + {"tracking_tag_id": "", + "timestamp_utc": datetime(2026, 1, 15, 10, 11, 30, tzinfo=timezone.utc), + "user_ids": [{"type": "DAID", "id": "a9342d1f-69f1-4bf8-bc2b-1f20eb451f21"}], + "order_id": "order-10045", "value": "59.98", "value_currency": "USD", + "event_name": "purchase"}, +] + +# schema= is required here: without it `user_ids` is inferred as MapType. +input_df = spark.createDataFrame(rows, schema=input_schema) +``` + +**Example notebook:** [Offline Conversion Data (CAPI) Example Notebook.ipynb](https://github.com/thetradedesk/ttd-databricks-python/blob/add-easy-start-examples-per-usecase-to-sdk-docs/example_notebook/Offline%20Conversion%20Data%20%28CAPI%29%20Example%20Notebook.ipynb). + +**Schema:** [offline_conversion.py](https://github.com/thetradedesk/ttd-databricks-python/blob/add-easy-start-examples-per-usecase-to-sdk-docs/ttd_databricks_python/ttd_databricks/schemas/offline_conversion.py). --- @@ -285,19 +434,37 @@ The input schema for `TTDEndpoint.OFFLINE_CONVERSION` is defined in [offline_con Deletion/Opt-Out endpoint scoped to a specific advertiser. ```python -from ttd_databricks_python.ttd_databricks import DeletionOptOutAdvertiserContext, PartnerDsrRequestType +from ttd_data.models import PartnerDsrRequestType + +from ttd_databricks_python.ttd_databricks import DeletionOptOutAdvertiserContext # request_type controls the action: # PartnerDsrRequestType.DELETION — remove user data # PartnerDsrRequestType.OPT_OUT — suppress future targeting context = DeletionOptOutAdvertiserContext( advertiser_id="", - request_type=PartnerDsrRequestType.OPT_OUT, # or OPT_OUT + request_type=PartnerDsrRequestType.OPT_OUT, # or DELETION data_provider_id="", # optional ) ``` -The input schema for `TTDEndpoint.DELETION_OPTOUT_ADVERTISER` is defined in [deletion_optout_advertiser.py](https://github.com/thetradedesk/ttd-databricks-python/blob/main/ttd_databricks_python/ttd_databricks/schemas/deletion_optout_advertiser.py). See the [Inspecting Schemas](#inspecting-schemas) section for helper functions to view columns and validate your DataFrame. +**Mandatory columns:** `id_type`, `id_value` — the only two columns this schema has. All three deletion/opt-out endpoints take the same input shape. + +```python +input_schema = get_ttd_input_schema(TTDEndpoint.DELETION_OPTOUT_ADVERTISER) + +rows = [ + {"id_type": "TDID", "id_value": "123e4567-e89b-12d3-a456-426652340000"}, + {"id_type": "DAID", "id_value": "a9342d1f-69f1-4bf8-bc2b-1f20eb451f21"}, + {"id_type": "UID2", "id_value": "48MjlfIUZpOKNAm9nod7/jCLAXUYsnE1tpVHQSDS0uo="}, +] + +input_df = spark.createDataFrame(rows, schema=input_schema) +``` + +**Example notebook:** [Deletion and Opt-Out (DSR) Example Notebook.ipynb](https://github.com/thetradedesk/ttd-databricks-python/blob/add-easy-start-examples-per-usecase-to-sdk-docs/example_notebook/Deletion%20and%20Opt-Out%20%28DSR%29%20Example%20Notebook.ipynb). + +**Schema:** [deletion_optout_advertiser.py](https://github.com/thetradedesk/ttd-databricks-python/blob/add-easy-start-examples-per-usecase-to-sdk-docs/ttd_databricks_python/ttd_databricks/schemas/deletion_optout_advertiser.py). --- @@ -306,16 +473,22 @@ The input schema for `TTDEndpoint.DELETION_OPTOUT_ADVERTISER` is defined in [del Deletion/Opt-Out endpoint scoped to a third-party data provider. ```python -from ttd_databricks_python.ttd_databricks import DeletionOptOutThirdPartyContext, PartnerDsrRequestType +from ttd_data.models import PartnerDsrRequestType + +from ttd_databricks_python.ttd_databricks import DeletionOptOutThirdPartyContext context = DeletionOptOutThirdPartyContext( data_provider_id="", - request_type=PartnerDsrRequestType.OPT_OUT, # or OPT_OUT + request_type=PartnerDsrRequestType.OPT_OUT, # or DELETION brand_id="", # optional ) ``` -The input schema for `TTDEndpoint.DELETION_OPTOUT_THIRDPARTY` is defined in [deletion_optout_thirdparty.py](https://github.com/thetradedesk/ttd-databricks-python/blob/main/ttd_databricks_python/ttd_databricks/schemas/deletion_optout_thirdparty.py). See the [Inspecting Schemas](#inspecting-schemas) section for helper functions to view columns and validate your DataFrame. +**Mandatory columns:** `id_type`, `id_value` — same input shape as the advertiser endpoint above. + +**Example notebook:** [Deletion and Opt-Out (DSR) Example Notebook.ipynb](https://github.com/thetradedesk/ttd-databricks-python/blob/add-easy-start-examples-per-usecase-to-sdk-docs/example_notebook/Deletion%20and%20Opt-Out%20%28DSR%29%20Example%20Notebook.ipynb). + +**Schema:** [deletion_optout_thirdparty.py](https://github.com/thetradedesk/ttd-databricks-python/blob/add-easy-start-examples-per-usecase-to-sdk-docs/ttd_databricks_python/ttd_databricks/schemas/deletion_optout_thirdparty.py). --- @@ -324,105 +497,133 @@ The input schema for `TTDEndpoint.DELETION_OPTOUT_THIRDPARTY` is defined in [del Deletion/Opt-Out endpoint scoped to a merchant. ```python -from ttd_databricks_python.ttd_databricks import DeletionOptOutMerchantContext, PartnerDsrRequestType +from ttd_data.models import PartnerDsrRequestType + +from ttd_databricks_python.ttd_databricks import DeletionOptOutMerchantContext context = DeletionOptOutMerchantContext( - merchant_id="", - request_type=PartnerDsrRequestType.OPT_OUT, # or OPT_OUT + merchant_id=123456, # int, not a string + request_type=PartnerDsrRequestType.OPT_OUT, # or DELETION ) ``` -The input schema for `TTDEndpoint.DELETION_OPTOUT_MERCHANT` is defined in [deletion_optout_merchant.py](https://github.com/thetradedesk/ttd-databricks-python/blob/main/ttd_databricks_python/ttd_databricks/schemas/deletion_optout_merchant.py). See the [Inspecting Schemas](#inspecting-schemas) section for helper functions to view columns and validate your DataFrame. +**Mandatory columns:** `id_type`, `id_value` — same input shape as the advertiser endpoint above. ---- +**Example notebook:** [Deletion and Opt-Out (DSR) Example Notebook.ipynb](https://github.com/thetradedesk/ttd-databricks-python/blob/add-easy-start-examples-per-usecase-to-sdk-docs/example_notebook/Deletion%20and%20Opt-Out%20%28DSR%29%20Example%20Notebook.ipynb). -### Inspecting Schemas +**Schema:** [deletion_optout_merchant.py](https://github.com/thetradedesk/ttd-databricks-python/blob/add-easy-start-examples-per-usecase-to-sdk-docs/ttd_databricks_python/ttd_databricks/schemas/deletion_optout_merchant.py). -Retrieve the full input schema for an endpoint: +--- -```python -from ttd_databricks_python.ttd_databricks import TTDEndpoint, get_ttd_input_schema +## Error Handling -schema = get_ttd_input_schema(TTDEndpoint.ADVERTISER) +All SDK exceptions inherit from `TTDError`. -for field in input_schema.fields: - print(f" {field.name}: {field.dataType.simpleString()} (nullable={field.nullable})") -``` +`TTDSchemaValidationError` and `TTDConfigurationError` are raised up front, before any data is sent. Schema validation and the Spark and Delta table checks all run ahead of the first API call. These propagate to you rather than being reported inline, so no DataFrame is returned and no rows reach The Trade Desk. + +Neither `push_data` nor `batch_process` raises an exception on API call failures. Both always return, including the rows that already succeeded, and every outcome is captured inline in the returned DataFrame via the `success`, `error_code`, and `error_message` columns. -Get just the required column names (useful for DataFrame preparation): +An unrecoverable error is an auth or permission failure (`401`/`403`), which would recur on every following call. When one occurs, the batch that hit it is recorded with that error's own code, and the remaining rows are recorded with `error_code="ABORTED"`. Those rows are never sent to The Trade Desk, so they are safe to re-run. Transient failures such as `429` and `5xx` are retried; if they still fail, they fail only their own batch and later batches carry on. ```python -from ttd_databricks_python.ttd_databricks.schemas import get_required_column_names +from ttd_databricks_python.ttd_databricks.exceptions import ( + TTDError, + TTDConfigurationError, + TTDSchemaValidationError, +) -required_cols = get_required_column_names(TTDEndpoint.ADVERTISER) -# e.g. ["id_type", "id_value", "segment_name"] +try: + result_df = client.push_data(df=input_df, context=context) +except TTDSchemaValidationError as e: + print(f"Missing columns: {e.missing_columns}") +except TTDConfigurationError as e: + print(f"Configuration error: {e}") ``` -Pre-validate a DataFrame before calling `push_data` to catch schema issues early: +| Exception | Cause | +|---|---| +| `TTDSchemaValidationError` | DataFrame is missing required columns for the endpoint | +| `TTDConfigurationError` | SparkSession not found, PySpark not installed, a required Delta table is missing, or an existing output table is missing expected columns | +| `TTDApiError` | A batch hit a failure no later batch could survive, such as an auth or permission error. Raised internally and caught by `push_data` and `batch_process`, which report it inline instead of propagating it | -```python -from ttd_databricks_python.ttd_databricks.schemas import validate_ttd_schema +--- -# Raises TTDSchemaValidationError if any required columns are missing. -validate_ttd_schema(df=input_df, endpoint=TTDEndpoint.ADVERTISER) -``` +## Optional Configuration ---- +None of the following is required to send data. Each endpoint already targets its own default server, retries are on by default, and UID2 resolution is only needed if you send raw email addresses or phone numbers. -## UID2 Support +The following are optional configurations clients can leverage to customize their integration. -The SDK supports both submitting UID2s directly as identifiers as well as automatically resolving raw email addresses and phone numbers (including pre-hashed variants) to UID2s before sending to The Trade Desk Data API. To enable automatic resolution, pass a `uid2_config` to `TtdDatabricksClient.from_params()`. +### UID2 Support -Email addresses and phone numbers can be submitted similar to other data types and are resolved to UID2s using the provided UID2 operator before calling The Trade Desk Data API. The Trade Desk Data API only receives resolved UID2s, never raw emails or phone numbers. Resolution happens per row in both `push_data` and `batch_process`, with UID2 mapping returned in a `uid2_resolutions` column on the output. +UID2s you have already resolved can be sent as-is, with `id_type` set to `UID2`. No configuration is needed for that. -To submit email addresses or phone numbers, set the `id_type` column in your input data to `Email`, `Phone`, `HashedEmail`, or `HashedPhone`, with the corresponding value in `id_value`. +The SDK can also resolve raw email addresses and phone numbers (including pre-hashed variants) to UID2s for you. Attach a `uid2_config` when you create the client: ---- +```python +from ttd_data.uid2 import IdentityScope, UID2Config +from ttd_databricks_python.ttd_databricks import TtdDatabricksClient -## Error Handling +uid2_config = UID2Config( + base_url="", + api_key="", + client_secret="", + identity_scope=IdentityScope.UID2, # use IdentityScope.EUID for European identities +) -All SDK exceptions inherit from `TTDError`. +client = TtdDatabricksClient.from_params( + api_token="", + uid2_config=uid2_config, +) +``` -Both `push_data` abd `batch_process` do not raise API call failures — a batch that hits an unrecoverable error fails with its error code, and rows succeeding that were never sent The Trade Desk and fail with `error_code="ABORTED"`. Both are captured inline in the result DataFrame via the `success`, `error_code`, and `error_message` columns, so processing is never interrupted by API or row-level failures. +Both client styles accept it. With dependency injection, pass it to the `DataClient` instead: ```python -from ttd_databricks_python.ttd_databricks.exceptions import ( - TTDError, - TTDConfigurationError, - TTDSchemaValidationError, +from ttd_data import DataClient +from ttd_databricks_python.ttd_databricks import TtdDatabricksClient + +client = TtdDatabricksClient( + data_api_client=DataClient(uid2_config=uid2_config), + api_token="", ) +``` -try: - result_df = client.push_data(df=input_df, context=context) -except TTDSchemaValidationError as e: - print(f"Missing columns: {e.missing_columns}") -except TTDConfigurationError as e: - print(f"Configuration error: {e}") +Then set `id_type` to `Email`, `Phone`, `HashedEmail`, or `HashedPhone` with the corresponding value in `id_value`. From there the rows go through `push_data` and `batch_process` like any other identifier type, and nothing else about your pipeline changes: + +```python +rows = [ + {"id_type": "Email", "id_value": "user@example.com", + "segment_name": "my_first_segment", "ttl_in_minutes": 43200}, + {"id_type": "HashedEmail", "id_value": "tMmiiTI7IaAcPpQPFQ65uMVCWH8av9jw4cwf/F5HVRQ=", + "segment_name": "my_first_segment", "ttl_in_minutes": 43200}, +] + +result_df = client.push_data(df=spark.createDataFrame(rows, schema=input_schema), context=context) + +result_df.select("id_type", "success", "error_message", "uid2_resolutions").show(truncate=False) ``` -| Exception | Cause | -|---|---| -| `TTDSchemaValidationError` | DataFrame is missing required columns for the endpoint | -| `TTDConfigurationError` | SparkSession not found or PySpark not installed | +Each identifier is resolved by your UID2 operator before the request leaves Databricks, so The Trade Desk only ever receives resolved UID2s, never raw emails or phone numbers. Resolution happens per row in both `push_data` and `batch_process`, and the raw-identifier-to-UID2 mapping comes back in the `uid2_resolutions` column. --- -## Server Selection +### Server Selection -Each endpoint has its own default server URL, sourced from the `ttd-data` SDK: +The following table shows the mapping between the context in `ttd-databricks`, the destination endpoint the data is sent to, and the default server used. The default servers are already set in the SDK, so you do not need to choose one. You do, however, have the flexibility to override the server that data is sent to. For the servers available to you, see [servers available for advertisers](https://open.thetradedesk.com/advertiser/docsApp/GuidesAdvertiser/data/doc/DataApiCallsAdvertiser#first-pd) and [servers available for data providers](https://open.thetradedesk.com/provider/docsApp/GuidesProvider/audience/doc/DataApiCallsProvider#third-pd) on OpenTTD. -| Endpoint | Path | Default Server | +| Context | Destination Endpoint | Default Server | |---|---|---| -| First-Party Data | `/data/advertiser` | `https://usw-data.adsrvr.org` | -| Third-Party Data | `/data/thirdparty` | `https://bulk-data.adsrvr.org` | -| Offline Conversion | `/providerapi/offlineconversion` | `https://offlineattrib.adsrvr.org` | -| Deletion / Opt-Out — Advertiser | `/data/deletion-optout/advertiser` | `https://usw-data.adsrvr.org` | -| Deletion / Opt-Out — Third Party | `/data/deletion-optout/thirdparty` | `https://usw-data.adsrvr.org` | -| Deletion / Opt-Out — Merchant | `/data/deletion-optout/merchant` | `https://usw-data.adsrvr.org` | +| `AdvertiserContext` | `POST /data/advertiser` | `https://usw-data.adsrvr.org` | +| `ThirdPartyContext` | `POST /data/thirdparty` | `https://bulk-data.adsrvr.org` | +| `OfflineConversionContext` | `POST /providerapi/offlineconversion` | `https://offlineattrib.adsrvr.org` | +| `DeletionOptOutAdvertiserContext` | `POST /data/deletion-optout/advertiser` | `https://usw-data.adsrvr.org` | +| `DeletionOptOutThirdPartyContext` | `POST /data/deletion-optout/thirdparty` | `https://usw-data.adsrvr.org` | +| `DeletionOptOutMerchantContext` | `POST /data/deletion-optout/merchant` | `https://usw-data.adsrvr.org` | -These can be overridden globally at the client level, or per-request via the context. +Override globally at the client level, or per request via the context: -### Global Override +#### Global Override Applies to all endpoints on the client: @@ -433,7 +634,7 @@ client = TtdDatabricksClient.from_params( ) ``` -### Per-Request Override +#### Per-Request Override Applies only to calls made with that context, leaving the client default unchanged for other endpoints: @@ -446,9 +647,9 @@ context = AdvertiserContext( --- -## Custom HTTP Client +### Custom HTTP Client -The underlying HTTP client is provided by the `ttd-data` SDK via [`DataClient`](https://github.com/thetradedesk/ttd-data-python/blob/main/src/ttd_data/sdk.py). You can inject a custom instance to configure the server URL or connection behaviour. +The underlying HTTP client is provided by the `ttd-data` SDK via [`DataClient`](https://github.com/thetradedesk/ttd-data-python/blob/main/src/ttd_data/sdk.py). You can inject a custom instance to configure the server URL or connection behaviour, or to inject a mock in tests. ```python from ttd_data import DataClient diff --git a/example_notebook/Deletion and Opt-Out (DSR) Example Notebook.ipynb b/example_notebook/Deletion and Opt-Out (DSR) Example Notebook.ipynb new file mode 100644 index 0000000..3c5c78b --- /dev/null +++ b/example_notebook/Deletion and Opt-Out (DSR) Example Notebook.ipynb @@ -0,0 +1,200 @@ +{ + "cells": [ + { + "cell_type": "markdown", + "metadata": {}, + "source": "# Deletion and Opt-Out (DSR) Example Notebook\n\nHonour a user's data subject request \u2014 delete their data, or opt them out of future\ntargeting \u2014 across every segment tied to your account.\n\n## Is this the right notebook for you?\n\n| | |\n|---|---|\n| **Who Should Use This?** | Anyone processing data subject requests: advertisers, data providers, merchants |\n| **What Does the Notebook Do?** | Deletes a user's data, or suppresses them from future targeting |\n| **What Data Does It Send?** | One row per identity \u2014 `id_type` and `id_value` only |\n| **Destination Trade Desk Endpoints** | The three `/data/deletion-optout/*` endpoints (via the `DeletionOptOutAdvertiserContext`, `DeletionOptOutThirdPartyContext` and `DeletionOptOutMerchantContext` classes of the ttd-databricks SDK) |\n\n### Deletion or opt-out?\n\nBoth actions use the same endpoints and the same input schema. Only `request_type` differs:\n\n| `request_type` | Effect |\n|---|---|\n| `PartnerDsrRequestType.OPT_OUT` | The user is suppressed from future targeting |\n| `PartnerDsrRequestType.DELETION` | The user's data is removed |\n\n### Which endpoint?\n\nPick the one matching the data you are acting on. Step 3 has a cell for each.\n\n| Your data | Context | Relevant OpenTTD API Documentation |\n|---|---|---|\n| First-party audience data | `DeletionOptOutAdvertiserContext` | [Advertiser](https://open.thetradedesk.com/advertiser/docsApp/GuidesAdvertiser/data/doc/post-data-deletion-optout-advertiser) \u00b7 [External provider](https://open.thetradedesk.com/provider/docsApp/GuidesProvider/audience/doc/post-data-deletion-optout-advertiser-external) |\n| Third-party audience data | `DeletionOptOutThirdPartyContext` | [Third party](https://open.thetradedesk.com/provider/docsApp/GuidesProvider/audience/doc/post-data-deletion-optout-thirdparty) |\n| Merchant data | `DeletionOptOutMerchantContext` | [Merchant](https://open.thetradedesk.com/provider/docsApp/GuidesProvider/retail/doc/post-data-deletion-optout-merchant) |\n\n## What you need\n\n- [ ] A TTD Platform API token \u2014 see [Create an API token](https://open.thetradedesk.com/advertiser/docsApp/GuidesAdvertiser/data/doc/DataApiCallsAdvertiser#ui-method-create)\n- [ ] The ID for your endpoint: advertiser ID, data provider ID, or merchant ID" + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": "## Step 1: Install the SDK" + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": "%pip install ttd-databricks\n\ndbutils.library.restartPython()" + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": "## Step 2: Configure credentials\n\nFor a first run you can paste values inline. In production, read them from Databricks Secrets:\n\n```python\nAPI_TOKEN = dbutils.secrets.get(scope=\"ttd\", key=\"api-token\")\nADVERTISER_ID = dbutils.secrets.get(scope=\"ttd\", key=\"advertiser-id\")\n```\n\n> **Note:** Authenticate with a Platform API token, sent as the `TTD-Auth` header. Generate one\n> in the OpenTTD Access Management app \u2014 see [Create an API token](https://open.thetradedesk.com/advertiser/docsApp/GuidesAdvertiser/data/doc/DataApiCallsAdvertiser#ui-method-create). Secret keys and\n> `TtdSignature` headers are a legacy method and are not supported by this SDK." + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": "API_TOKEN = \"\"\n\n# Set whichever applies to the endpoint you are using in Step 3.\nADVERTISER_ID = \"\"\nDATA_PROVIDER_ID = \"\"\nMERCHANT_ID = 0 # integer, not a string" + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": "## Step 3: Create the client and context\n\nRun **one** of the three cells below \u2014 whichever matches your data. Each sets `context`\nand `ENDPOINT`; every later step works the same way regardless of which you chose.\n\n`PartnerDsrRequestType` comes from `ttd_data.models`, not from the `ttd_databricks` package." + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": "from ttd_data.models import PartnerDsrRequestType\n\nfrom ttd_databricks_python.ttd_databricks import (\n TTDEndpoint,\n TtdDatabricksClient,\n get_ttd_input_schema,\n)\n\n# Two ways to create the client \u2014 this notebook uses (ii) everywhere below.\n# i) Dependency injection \u2014 you build the DataClient and pass it in:\n# from ttd_data import DataClient\n# client = TtdDatabricksClient(data_api_client=DataClient(), api_token=API_TOKEN)\n# ii) Factory \u2014 from_params builds the DataClient for you:\nclient = TtdDatabricksClient.from_params(api_token=API_TOKEN)\n\n# Switch to PartnerDsrRequestType.DELETION to delete instead of opt out.\nREQUEST_TYPE = PartnerDsrRequestType.OPT_OUT" + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": "### Option A \u2014 first-party (advertiser) data" + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": "from ttd_databricks_python.ttd_databricks import TTDEndpoint, DeletionOptOutAdvertiserContext\n\nENDPOINT = TTDEndpoint.DELETION_OPTOUT_ADVERTISER\ncontext = DeletionOptOutAdvertiserContext(\n advertiser_id=ADVERTISER_ID,\n request_type=REQUEST_TYPE,\n data_provider_id=None, # optional; set only if you are an external provider\n)\n\nprint(f\"Context: {context}\")" + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": "### Option B \u2014 third-party (data provider) data" + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": "from ttd_databricks_python.ttd_databricks import TTDEndpoint, DeletionOptOutThirdPartyContext\n\nENDPOINT = TTDEndpoint.DELETION_OPTOUT_THIRDPARTY\ncontext = DeletionOptOutThirdPartyContext(\n data_provider_id=DATA_PROVIDER_ID,\n request_type=REQUEST_TYPE,\n brand_id=None, # optional\n)\n\nprint(f\"Context: {context}\")" + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": "### Option C \u2014 merchant data" + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": "from ttd_databricks_python.ttd_databricks import TTDEndpoint, DeletionOptOutMerchantContext\n\nENDPOINT = TTDEndpoint.DELETION_OPTOUT_MERCHANT\ncontext = DeletionOptOutMerchantContext(\n merchant_id=MERCHANT_ID, # integer\n request_type=REQUEST_TYPE,\n)\n\nprint(f\"Context: {context}\")" + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": "## Step 4: Inspect the required input schema\n\nAll three endpoints take the same two mandatory columns and nothing else:\n`id_type` and `id_value`. There is no segment or timestamp \u2014 the request applies across\nevery segment tied to your account." + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": "from ttd_databricks_python.ttd_databricks import get_ttd_input_schema\nfrom ttd_databricks_python.ttd_databricks.schemas import get_required_column_names\n\ninput_schema = get_ttd_input_schema(ENDPOINT)\n\nprint(\"Mandatory columns:\", get_required_column_names(ENDPOINT))\nprint(\"\\nFull input schema:\")\nfor field in input_schema.fields:\n print(f\" {field.name}: {field.dataType.simpleString()} (nullable={field.nullable})\")" + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": "## Step 5: Prepare your input DataFrame\n\nOne row per identity to delete or opt out.\n\n> **Tip:** Start with a handful of rows. These requests are not reversible, so confirming\n> the flow on a small sample matters more here than anywhere else." + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": "rows = [\n {\"id_type\": \"TDID\", \"id_value\": \"123e4567-e89b-12d3-a456-426652340000\"},\n {\"id_type\": \"DAID\", \"id_value\": \"a9342d1f-69f1-4bf8-bc2b-1f20eb451f21\"},\n {\"id_type\": \"UID2\", \"id_value\": \"48MjlfIUZpOKNAm9nod7/jCLAXUYsnE1tpVHQSDS0uo=\"},\n {\"id_type\": \"RampID\", \"id_value\": \"XY1005wXyWPB1SgpMUKIpzA0I3UaLEz-2lg0wFAr1PWK7FMhs\"},\n]\n\ninput_df = spark.createDataFrame(rows, schema=input_schema)\ndisplay(input_df)" + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": "## Step 6: Sending Data\n\nThere are two ways to send your data. Pick one \u2014 you do not need both.\n\n| | Ad hoc | Batch processing |\n|---|---|---|\n| **State management** | None. Every call sends every row you give it. | Provided. A metadata table records progress, so each run sends only rows added since the last one. |\n| **Input** | A DataFrame you build in the notebook | A Delta input table |\n| **Output** | Returned inline as a DataFrame | Written to a Delta output table |\n| **Best for** | One-off loads and first tests | Recurring pipelines |" + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": "### Step 6a: Ad Hoc Usage (No State Management Provided)\n\n`push_data` sends your DataFrame straight to the Data API and returns the input columns\nenriched with per-row status. The SDK keeps no record of what it has already sent, so\nre-running this cell sends every row again.\n\nIt does not raise on API or row-level failures \u2014 every outcome is reported inline." + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": "result_df = client.push_data(df=input_df, context=context, batch_size=1600)\n\ndisplay(result_df)" + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": "`push_data` adds these columns to your input:\n\n| Column | Meaning |\n|---|---|\n| `success` | `True` if the row was accepted |\n| `error_code` | Failure category, `null` on success |\n| `error_message` | Human-readable reason, `null` on success |\n| `processed_timestamp` | When the row was submitted |\n| `uid2_resolutions` | Raw identifier \u2192 UID2 mapping, empty unless `uid2_config` was set |" + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": "from pyspark.sql.functions import col\n\ntotal = result_df.count()\nsucceeded = result_df.filter(col(\"success\")).count()\n\nprint(f\"Total: {total} | Succeeded: {succeeded} | Failed: {total - succeeded}\")\n\nfailed_df = result_df.filter(~col(\"success\"))\nif failed_df.count():\n display(failed_df.select(\"error_code\", \"error_message\"))" + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": "### Step 6b: Batch Processing (State Management Provided)\n\n`batch_process` reads from a Delta input table, writes results to an output table, and\nrecords how far it got in a metadata table. With `process_new_records_only=True`, each run\npicks up only the rows added since the last successful run, so you can schedule it without\nre-sending history.\n\n**One time steps:** Create the three Delta tables. Every future run reuses these same tables \u2014\nthe metadata table is what remembers your progress, so do not drop or recreate it between\nruns. The `setup_*` methods are safe to re-run: they return the existing table if it is\nalready there." + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": "input_table = client.setup_input_table(endpoint=ENDPOINT)\noutput_table = client.setup_output_table(endpoint=ENDPOINT)\nmetadata_table = client.setup_metadata_table()\n\nprint(f\"Input table: {input_table}\")\nprint(f\"Output table: {output_table}\")\nprint(f\"Metadata table: {metadata_table}\")" + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": "**Every run:**\n\n**1. Append new rows to the input table.** In production this is your upstream pipeline's job." + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": "from pyspark.sql import functions as F\n\n(\n spark.createDataFrame(rows, schema=input_schema)\n .withColumn(\"updated_at\", F.current_timestamp())\n .write.format(\"delta\").mode(\"append\").saveAsTable(input_table)\n)\n\ndisplay(spark.table(input_table))" + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": "**2. Call `batch_process`.** Re-running it picks up only rows appended since the last run." + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": "client.batch_process(\n context=context,\n input_table=input_table,\n output_table=output_table,\n metadata_table=metadata_table,\n process_new_records_only=True, # incremental; set False to reprocess every row\n batch_size=1600, # rows per API request\n)\n\ndisplay(spark.table(output_table))\ndisplay(spark.table(metadata_table))" + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": "## (Optional) Sending email addresses or phone numbers (UID2)\n\nSkip this section if you are sending device IDs or UID2s you have already resolved.\n\nPass a `uid2_config` when you create the client. It is the same call as Step 3 with one\nextra argument. Then set `id_type` to `Email`, `Phone`,\n`HashedEmail`, or `HashedPhone` on your rows. From there `push_data` and `batch_process` take them exactly like any other\nidentifier type, and nothing else about Step 6a or Step 6b changes.\n\nThe SDK resolves each identifier to a UID2 (or EUID) using your operator before the\nrequest leaves Databricks, so The Trade Desk never receives the raw email or phone\nnumber. The mapping comes back in the `uid2_resolutions` column." + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": "from ttd_data.uid2 import IdentityScope, UID2Config\n\nfrom ttd_databricks_python.ttd_databricks import TtdDatabricksClient\n\nuid2_client = TtdDatabricksClient.from_params(\n api_token=API_TOKEN,\n uid2_config=UID2Config(\n base_url=\"\",\n api_key=\"\",\n client_secret=\"\",\n identity_scope=IdentityScope.UID2, # use IdentityScope.EUID for European identities\n ),\n)\n\nuid2_data = [\n {\"id_type\": \"Email\", \"id_value\": \"user@example.com\"},\n {\"id_type\": \"HashedEmail\", \"id_value\": \"tMmiiTI7IaAcPpQPFQ65uMVCWH8av9jw4cwf/F5HVRQ=\"},\n]\n\nuid2_result_df = uid2_client.push_data(\n df=spark.createDataFrame(uid2_data, schema=input_schema),\n context=context,\n)\n\ndisplay(uid2_result_df.select(\"id_type\", \"success\", \"error_message\", \"uid2_resolutions\"))" + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": "## Next steps\n\n- **Other use cases** \u2014 one notebook per use case:\n [First Party Data (1PD) Example Notebook](https://github.com/thetradedesk/ttd-databricks-python/blob/add-easy-start-examples-per-usecase-to-sdk-docs/example_notebook/First%20Party%20Data%20%281PD%29%20Example%20Notebook.ipynb),\n [Third Party Data (3PD) Example Notebook](https://github.com/thetradedesk/ttd-databricks-python/blob/add-easy-start-examples-per-usecase-to-sdk-docs/example_notebook/Third%20Party%20Data%20%283PD%29%20Example%20Notebook.ipynb),\n [Offline Conversion Data (CAPI) Example Notebook](https://github.com/thetradedesk/ttd-databricks-python/blob/add-easy-start-examples-per-usecase-to-sdk-docs/example_notebook/Offline%20Conversion%20Data%20%28CAPI%29%20Example%20Notebook.ipynb),\n [Deletion and Opt-Out (DSR) Example Notebook](https://github.com/thetradedesk/ttd-databricks-python/blob/add-easy-start-examples-per-usecase-to-sdk-docs/example_notebook/Deletion%20and%20Opt-Out%20%28DSR%29%20Example%20Notebook.ipynb).\n- **Full reference** \u2014 [README](https://github.com/thetradedesk/ttd-databricks-python/blob/add-easy-start-examples-per-usecase-to-sdk-docs/README.md) covers authentication, error handling,\n UID2 support, custom HTTP clients, and server URL overrides.\n- **Server URLs** \u2014 you do not need to configure one. Each endpoint already targets its\n own default server, and it can be overridden with a preferred server if you need one." + } + ], + "metadata": { + "kernelspec": { + "display_name": "Python 3", + "language": "python", + "name": "python3" + }, + "language_info": { + "name": "python", + "version": "3.10.0" + } + }, + "nbformat": 4, + "nbformat_minor": 4 +} diff --git a/example_notebook/First Party Data (1PD) Example Notebook.ipynb b/example_notebook/First Party Data (1PD) Example Notebook.ipynb new file mode 100644 index 0000000..cc37854 --- /dev/null +++ b/example_notebook/First Party Data (1PD) Example Notebook.ipynb @@ -0,0 +1,164 @@ +{ + "cells": [ + { + "cell_type": "markdown", + "metadata": {}, + "source": "# First Party Data (1PD) Example Notebook\n\nSend your own customer audiences from Databricks to The Trade Desk for targeting.\n\n## Is this the right notebook for you?\n\n| | |\n|---|---|\n| **Who Should Use This?** | Advertisers uploading their own customer data, and onboarders acting on an advertiser's behalf |\n| **What Does the Notebook Do?** | Adds identities to a first-party audience segment so you can target them |\n| **What Data Does It Send?** | One row per identity per segment |\n| **Destination Trade Desk Endpoint** | `POST /data/advertiser` (via the `AdvertiserContext` class of the ttd-databricks SDK) |\n| **Relevant OpenTTD API Documentation** | [First-party data](https://open.thetradedesk.com/advertiser/docsApp/GuidesAdvertiser/data/doc/post-data-advertiser-firstparty) \u00b7 [External provider](https://open.thetradedesk.com/provider/docsApp/GuidesProvider/audience/doc/post-data-advertiser-external) |\n\nOnboarding conversion events instead? Use [Offline Conversion Data (CAPI) Example Notebook](https://github.com/thetradedesk/ttd-databricks-python/blob/add-easy-start-examples-per-usecase-to-sdk-docs/example_notebook/Offline%20Conversion%20Data%20%28CAPI%29%20Example%20Notebook.ipynb).\nRemoving users? Use [Deletion and Opt-Out (DSR) Example Notebook](https://github.com/thetradedesk/ttd-databricks-python/blob/add-easy-start-examples-per-usecase-to-sdk-docs/example_notebook/Deletion%20and%20Opt-Out%20%28DSR%29%20Example%20Notebook.ipynb).\n\n## What you need\n\n- [ ] A TTD Platform API token \u2014 see [Create an API token](https://open.thetradedesk.com/advertiser/docsApp/GuidesAdvertiser/data/doc/DataApiCallsAdvertiser#ui-method-create)\n- [ ] Your advertiser ID\n- [ ] (Optional) A data provider ID \u2014 only if you are an external provider acting for the advertiser\n- [ ] (Optional) A UID2 operator base URL, API key and client secret \u2014 to send raw email addresses or phone numbers" + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": "## Step 1: Install the SDK" + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": "%pip install ttd-databricks\n\ndbutils.library.restartPython()" + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": "## Step 2: Configure credentials\n\nFor a first run you can paste values inline. In production, read them from Databricks Secrets:\n\n```python\nAPI_TOKEN = dbutils.secrets.get(scope=\"ttd\", key=\"api-token\")\nADVERTISER_ID = dbutils.secrets.get(scope=\"ttd\", key=\"advertiser-id\")\n```\n\n> **Note:** Authenticate with a Platform API token, sent as the `TTD-Auth` header. Generate one\n> in the OpenTTD Access Management app \u2014 see [Create an API token](https://open.thetradedesk.com/advertiser/docsApp/GuidesAdvertiser/data/doc/DataApiCallsAdvertiser#ui-method-create). Secret keys and\n> `TtdSignature` headers are a legacy method and are not supported by this SDK." + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": "API_TOKEN = \"\"\nADVERTISER_ID = \"\"\nDATA_PROVIDER_ID = None # optional; set only if you are an external provider" + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": "## Step 3: Create the client and context" + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": "from ttd_databricks_python.ttd_databricks import (\n AdvertiserContext,\n TTDEndpoint,\n TtdDatabricksClient,\n get_ttd_input_schema,\n)\n\n# Two ways to create the client \u2014 this notebook uses (ii) everywhere below.\n# i) Dependency injection \u2014 you build the DataClient and pass it in:\n# from ttd_data import DataClient\n# client = TtdDatabricksClient(data_api_client=DataClient(), api_token=API_TOKEN)\n# ii) Factory \u2014 from_params builds the DataClient for you:\nclient = TtdDatabricksClient.from_params(api_token=API_TOKEN)\n\ncontext = AdvertiserContext(\n advertiser_id=ADVERTISER_ID,\n data_provider_id=DATA_PROVIDER_ID,\n)\n\nprint(f\"Context: {context}\")" + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": "## Step 4: Inspect the required input schema\n\nMandatory columns: `id_type`, `id_value`, `segment_name`\n\nOptional columns: `cookie_mapping_partner_id`, `timestamp_utc`, `ttl_in_minutes`, `base_bid_cpm`, `base_bid_cpm_metadata`, `bid_factor`" + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": "from ttd_databricks_python.ttd_databricks import TTDEndpoint, get_ttd_input_schema\nfrom ttd_databricks_python.ttd_databricks.schemas import get_required_column_names\n\ninput_schema = get_ttd_input_schema(TTDEndpoint.ADVERTISER)\n\nprint(\"Mandatory columns:\", get_required_column_names(TTDEndpoint.ADVERTISER))\nprint(\"\\nFull input schema:\")\nfor field in input_schema.fields:\n print(f\" {field.name}: {field.dataType.simpleString()} (nullable={field.nullable})\")" + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": "## Step 5: Prepare your input DataFrame\n\nOne row per identity per segment. `segment_name` is the audience the identity joins, and\n`ttl_in_minutes` controls how long it stays targetable (default 90 days, maximum 180 days).\n\n> **Tip:** Start with a handful of rows. Confirming the end-to-end flow on a small sample is much easier to troubleshoot than a full load." + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": "from datetime import datetime, timezone\n\nrows = [\n {\"id_type\": \"TDID\", \"id_value\": \"123e4567-e89b-12d3-a456-426652340000\",\n \"segment_name\": \"my_first_segment\", \"ttl_in_minutes\": 43200},\n {\"id_type\": \"DAID\", \"id_value\": \"a9342d1f-69f1-4bf8-bc2b-1f20eb451f21\",\n \"segment_name\": \"my_first_segment\", \"ttl_in_minutes\": 43200},\n {\"id_type\": \"UID2\", \"id_value\": \"48MjlfIUZpOKNAm9nod7/jCLAXUYsnE1tpVHQSDS0uo=\",\n \"segment_name\": \"my_first_segment\", \"ttl_in_minutes\": 43200},\n {\"id_type\": \"RampID\", \"id_value\": \"XY1005wXyWPB1SgpMUKIpzA0I3UaLEz-2lg0wFAr1PWK7FMhs\",\n \"segment_name\": \"my_first_segment\", \"ttl_in_minutes\": 43200,\n \"timestamp_utc\": datetime(2026, 1, 15, 10, 0, 0, tzinfo=timezone.utc)},\n]\n\ninput_df = spark.createDataFrame(rows, schema=input_schema)\ndisplay(input_df)" + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": "## Step 6: Sending Data\n\nThere are two ways to send your data. Pick one \u2014 you do not need both.\n\n| | Ad hoc | Batch processing |\n|---|---|---|\n| **State management** | None. Every call sends every row you give it. | Provided. A metadata table records progress, so each run sends only rows added since the last one. |\n| **Input** | A DataFrame you build in the notebook | A Delta input table |\n| **Output** | Returned inline as a DataFrame | Written to a Delta output table |\n| **Best for** | One-off loads and first tests | Recurring pipelines |" + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": "### Step 6a: Ad Hoc Usage (No State Management Provided)\n\n`push_data` sends your DataFrame straight to the Data API and returns the input columns\nenriched with per-row status. The SDK keeps no record of what it has already sent, so\nre-running this cell sends every row again.\n\nIt does not raise on API or row-level failures \u2014 every outcome is reported inline." + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": "result_df = client.push_data(df=input_df, context=context, batch_size=1600)\n\ndisplay(result_df)" + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": "`push_data` adds these columns to your input:\n\n| Column | Meaning |\n|---|---|\n| `success` | `True` if the row was accepted |\n| `error_code` | Failure category, `null` on success |\n| `error_message` | Human-readable reason, `null` on success |\n| `processed_timestamp` | When the row was submitted |\n| `uid2_resolutions` | Raw identifier \u2192 UID2 mapping, empty unless `uid2_config` was set |" + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": "from pyspark.sql.functions import col\n\ntotal = result_df.count()\nsucceeded = result_df.filter(col(\"success\")).count()\n\nprint(f\"Total: {total} | Succeeded: {succeeded} | Failed: {total - succeeded}\")\n\nfailed_df = result_df.filter(~col(\"success\"))\nif failed_df.count():\n display(failed_df.select(\"error_code\", \"error_message\"))" + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": "### Step 6b: Batch Processing (State Management Provided)\n\n`batch_process` reads from a Delta input table, writes results to an output table, and\nrecords how far it got in a metadata table. With `process_new_records_only=True`, each run\npicks up only the rows added since the last successful run, so you can schedule it without\nre-sending history.\n\n**One time steps:** Create the three Delta tables. Every future run reuses these same tables \u2014\nthe metadata table is what remembers your progress, so do not drop or recreate it between\nruns. The `setup_*` methods are safe to re-run: they return the existing table if it is\nalready there." + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": "from ttd_databricks_python.ttd_databricks import TTDEndpoint\n\ninput_table = client.setup_input_table(endpoint=TTDEndpoint.ADVERTISER)\noutput_table = client.setup_output_table(endpoint=TTDEndpoint.ADVERTISER)\nmetadata_table = client.setup_metadata_table()\n\nprint(f\"Input table: {input_table}\")\nprint(f\"Output table: {output_table}\")\nprint(f\"Metadata table: {metadata_table}\")" + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": "**Every run:**\n\n**1. Append new rows to the input table.** In production this is your upstream pipeline's job." + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": "from pyspark.sql import functions as F\n\n(\n spark.createDataFrame(rows, schema=input_schema)\n .withColumn(\"updated_at\", F.current_timestamp())\n .write.format(\"delta\").mode(\"append\").saveAsTable(input_table)\n)\n\ndisplay(spark.table(input_table))" + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": "**2. Call `batch_process`.** Re-running it picks up only rows appended since the last run." + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": "client.batch_process(\n context=context,\n input_table=input_table,\n output_table=output_table,\n metadata_table=metadata_table,\n process_new_records_only=True, # incremental; set False to reprocess every row\n batch_size=1600, # rows per API request\n)\n\ndisplay(spark.table(output_table))\ndisplay(spark.table(metadata_table))" + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": "## (Optional) Sending email addresses or phone numbers (UID2)\n\nSkip this section if you are sending device IDs or UID2s you have already resolved.\n\nPass a `uid2_config` when you create the client. It is the same call as Step 3 with one\nextra argument. Then set `id_type` to `Email`, `Phone`,\n`HashedEmail`, or `HashedPhone` on your rows. From there `push_data` and `batch_process` take them exactly like any other\nidentifier type, and nothing else about Step 6a or Step 6b changes.\n\nThe SDK resolves each identifier to a UID2 (or EUID) using your operator before the\nrequest leaves Databricks, so The Trade Desk never receives the raw email or phone\nnumber. The mapping comes back in the `uid2_resolutions` column." + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": "from ttd_data.uid2 import IdentityScope, UID2Config\n\nfrom ttd_databricks_python.ttd_databricks import TtdDatabricksClient\n\nuid2_client = TtdDatabricksClient.from_params(\n api_token=API_TOKEN,\n uid2_config=UID2Config(\n base_url=\"\",\n api_key=\"\",\n client_secret=\"\",\n identity_scope=IdentityScope.UID2, # use IdentityScope.EUID for European identities\n ),\n)\n\nuid2_data = [\n {\"id_type\": \"Email\", \"id_value\": \"user@example.com\",\n \"segment_name\": \"my_first_segment\", \"ttl_in_minutes\": 43200},\n {\"id_type\": \"HashedEmail\", \"id_value\": \"tMmiiTI7IaAcPpQPFQ65uMVCWH8av9jw4cwf/F5HVRQ=\",\n \"segment_name\": \"my_first_segment\", \"ttl_in_minutes\": 43200},\n]\n\nuid2_result_df = uid2_client.push_data(\n df=spark.createDataFrame(uid2_data, schema=input_schema),\n context=context,\n)\n\ndisplay(uid2_result_df.select(\"id_type\", \"success\", \"error_message\", \"uid2_resolutions\"))" + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": "## Next steps\n\n- **Other use cases** \u2014 one notebook per use case:\n [First Party Data (1PD) Example Notebook](https://github.com/thetradedesk/ttd-databricks-python/blob/add-easy-start-examples-per-usecase-to-sdk-docs/example_notebook/First%20Party%20Data%20%281PD%29%20Example%20Notebook.ipynb),\n [Third Party Data (3PD) Example Notebook](https://github.com/thetradedesk/ttd-databricks-python/blob/add-easy-start-examples-per-usecase-to-sdk-docs/example_notebook/Third%20Party%20Data%20%283PD%29%20Example%20Notebook.ipynb),\n [Offline Conversion Data (CAPI) Example Notebook](https://github.com/thetradedesk/ttd-databricks-python/blob/add-easy-start-examples-per-usecase-to-sdk-docs/example_notebook/Offline%20Conversion%20Data%20%28CAPI%29%20Example%20Notebook.ipynb),\n [Deletion and Opt-Out (DSR) Example Notebook](https://github.com/thetradedesk/ttd-databricks-python/blob/add-easy-start-examples-per-usecase-to-sdk-docs/example_notebook/Deletion%20and%20Opt-Out%20%28DSR%29%20Example%20Notebook.ipynb).\n- **Full reference** \u2014 [README](https://github.com/thetradedesk/ttd-databricks-python/blob/add-easy-start-examples-per-usecase-to-sdk-docs/README.md) covers authentication, error handling,\n UID2 support, custom HTTP clients, and server URL overrides.\n- **Server URLs** \u2014 you do not need to configure one. Each endpoint already targets its\n own default server, and it can be overridden with a preferred server if you need one." + } + ], + "metadata": { + "kernelspec": { + "display_name": "Python 3", + "language": "python", + "name": "python3" + }, + "language_info": { + "name": "python", + "version": "3.10.0" + } + }, + "nbformat": 4, + "nbformat_minor": 4 +} diff --git a/example_notebook/Offline Conversion Data (CAPI) Example Notebook.ipynb b/example_notebook/Offline Conversion Data (CAPI) Example Notebook.ipynb new file mode 100644 index 0000000..8eaaab5 --- /dev/null +++ b/example_notebook/Offline Conversion Data (CAPI) Example Notebook.ipynb @@ -0,0 +1,164 @@ +{ + "cells": [ + { + "cell_type": "markdown", + "metadata": {}, + "source": "# Offline Conversion Data (CAPI) Example Notebook\n\nSend conversion events that happened away from your site \u2014 in store, in a call centre, in\na CRM \u2014 to The Trade Desk for campaign measurement and attribution.\n\n## Is this the right notebook for you?\n\n| | |\n|---|---|\n| **Who Should Use This?** | Advertisers, data providers, and commerce partners measuring offline conversions |\n| **What Does the Notebook Do?** | Attributes conversion events back to TTD campaigns |\n| **What Data Does It Send?** | One row per conversion **event** \u2014 not per audience membership |\n| **Destination Trade Desk Endpoint** | `POST /providerapi/offlineconversion` (via the `OfflineConversionContext` class of the ttd-databricks SDK) |\n| **Relevant OpenTTD API Documentation** | [Offline conversion](https://open.thetradedesk.com/advertiser/docsApp/GuidesAdvertiser/data/doc/post-providerapi-offlineconversion) |\n\nSending real-time, on-site conversions instead? That goes through the Real-Time Conversion\nEvents API, not this SDK.\n\n## What you need\n\n- [ ] A TTD Platform API token \u2014 see [Create an API token](https://open.thetradedesk.com/advertiser/docsApp/GuidesAdvertiser/data/doc/DataApiCallsAdvertiser#ui-method-create)\n- [ ] Your data provider ID\n- [ ] A tracking tag ID for the conversion you are reporting\n\n> **Note:** Identity works differently here. Instead of flat `id_type` / `id_value`\n> columns, each row carries a `user_ids` array of `{type, id}` structs \u2014 up to 20 per\n> event, so you can supply several identifiers for the same conversion." + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": "## Step 1: Install the SDK" + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": "%pip install ttd-databricks\n\ndbutils.library.restartPython()" + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": "## Step 2: Configure credentials\n\nFor a first run you can paste values inline. In production, read them from Databricks Secrets:\n\n```python\nAPI_TOKEN = dbutils.secrets.get(scope=\"ttd\", key=\"api-token\")\nDATA_PROVIDER_ID = dbutils.secrets.get(scope=\"ttd\", key=\"data-provider-id\")\nTRACKING_TAG_ID = dbutils.secrets.get(scope=\"ttd\", key=\"tracking-tag-id\")\n```\n\n> **Note:** Authenticate with a Platform API token, sent as the `TTD-Auth` header. Generate one\n> in the OpenTTD Access Management app \u2014 see [Create an API token](https://open.thetradedesk.com/advertiser/docsApp/GuidesAdvertiser/data/doc/DataApiCallsAdvertiser#ui-method-create). Secret keys and\n> `TtdSignature` headers are a legacy method and are not supported by this SDK." + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": "API_TOKEN = \"\"\nDATA_PROVIDER_ID = \"\"\nTRACKING_TAG_ID = \"\"" + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": "## Step 3: Create the client and context" + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": "from ttd_databricks_python.ttd_databricks import (\n OfflineConversionContext,\n TTDEndpoint,\n TtdDatabricksClient,\n get_ttd_input_schema,\n)\n\n# Two ways to create the client \u2014 this notebook uses (ii) everywhere below.\n# i) Dependency injection \u2014 you build the DataClient and pass it in:\n# from ttd_data import DataClient\n# client = TtdDatabricksClient(data_api_client=DataClient(), api_token=API_TOKEN)\n# ii) Factory \u2014 from_params builds the DataClient for you:\nclient = TtdDatabricksClient.from_params(api_token=API_TOKEN)\n\ncontext = OfflineConversionContext(data_provider_id=DATA_PROVIDER_ID)\n\nprint(f\"Context: {context}\")" + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": "## Step 4: Inspect the required input schema\n\nMandatory columns: `tracking_tag_id`, `timestamp_utc`\n\nOptional columns: `user_ids`, `order_id`, `value`, `value_currency`, `event_name`, `merchant_id`, `impression_id`, `country`, `region`, `metro`, `city`, `line_items`, `privacy_settings`, `td1` \u2026 `td10`" + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": "from ttd_databricks_python.ttd_databricks import TTDEndpoint, get_ttd_input_schema\nfrom ttd_databricks_python.ttd_databricks.schemas import get_required_column_names\n\ninput_schema = get_ttd_input_schema(TTDEndpoint.OFFLINE_CONVERSION)\n\nprint(\"Mandatory columns:\", get_required_column_names(TTDEndpoint.OFFLINE_CONVERSION))\nprint(\"\\nFull input schema:\")\nfor field in input_schema.fields:\n print(f\" {field.name}: {field.dataType.simpleString()} (nullable={field.nullable})\")" + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": "## Step 5: Prepare your input DataFrame\n\n`user_ids` is optional in the schema but you must supply **either** `user_ids` **or**\n`impression_id` \u2014 a conversion with no identity cannot be attributed.\n\n> **Important:** pass `schema=input_schema` to `spark.createDataFrame`. Without it Spark\n> infers `MapType` for the nested `user_ids`, `line_items`, and `privacy_settings` columns\n> and the SDK cannot read them.\n\n> **Tip:** Start with a handful of rows. Confirming the end-to-end flow on a small sample is much easier to troubleshoot than a full load." + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": "from datetime import datetime, timezone\n\nTS = datetime(2026, 1, 15, 10, 11, 30, tzinfo=timezone.utc)\n\nrows = [\n {\"tracking_tag_id\": TRACKING_TAG_ID, \"timestamp_utc\": TS,\n \"user_ids\": [{\"type\": \"TDID\", \"id\": \"123e4567-e89b-12d3-a456-426652340000\"}]},\n {\"tracking_tag_id\": TRACKING_TAG_ID, \"timestamp_utc\": TS,\n \"user_ids\": [\n {\"type\": \"DAID\", \"id\": \"a9342d1f-69f1-4bf8-bc2b-1f20eb451f21\"},\n {\"type\": \"UID2\", \"id\": \"48MjlfIUZpOKNAm9nod7/jCLAXUYsnE1tpVHQSDS0uo=\"},\n ]},\n {\"tracking_tag_id\": TRACKING_TAG_ID, \"timestamp_utc\": TS,\n \"user_ids\": [{\"type\": \"TDID\", \"id\": \"1364cfc8-8850-4fe4-93f2-8728ed4d0b2d\"}],\n \"order_id\": \"order-10045\", \"value\": \"59.98\", \"value_currency\": \"USD\",\n \"event_name\": \"purchase\",\n \"line_items\": [\n {\"item_code\": \"203319203\", \"name\": \"Blue T-Shirt\", \"qty\": \"2\",\n \"price\": \"29.99\", \"cat\": \"Clothes\"},\n ]},\n]\n\ninput_df = spark.createDataFrame(rows, schema=input_schema)\ndisplay(input_df)" + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": "## Step 6: Sending Data\n\nThere are two ways to send your data. Pick one \u2014 you do not need both.\n\n| | Ad hoc | Batch processing |\n|---|---|---|\n| **State management** | None. Every call sends every row you give it. | Provided. A metadata table records progress, so each run sends only rows added since the last one. |\n| **Input** | A DataFrame you build in the notebook | A Delta input table |\n| **Output** | Returned inline as a DataFrame | Written to a Delta output table |\n| **Best for** | One-off loads and first tests | Recurring pipelines |" + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": "### Step 6a: Ad Hoc Usage (No State Management Provided)\n\n`push_data` sends your DataFrame straight to the Data API and returns the input columns\nenriched with per-row status. The SDK keeps no record of what it has already sent, so\nre-running this cell sends every row again.\n\nIt does not raise on API or row-level failures \u2014 every outcome is reported inline." + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": "result_df = client.push_data(df=input_df, context=context, batch_size=1600)\n\ndisplay(result_df)" + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": "`push_data` adds these columns to your input:\n\n| Column | Meaning |\n|---|---|\n| `success` | `True` if the row was accepted |\n| `error_code` | Failure category, `null` on success |\n| `error_message` | Human-readable reason, `null` on success |\n| `processed_timestamp` | When the row was submitted |\n| `uid2_resolutions` | Raw identifier \u2192 UID2 mapping, empty unless `uid2_config` was set |" + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": "from pyspark.sql.functions import col\n\ntotal = result_df.count()\nsucceeded = result_df.filter(col(\"success\")).count()\n\nprint(f\"Total: {total} | Succeeded: {succeeded} | Failed: {total - succeeded}\")\n\nfailed_df = result_df.filter(~col(\"success\"))\nif failed_df.count():\n display(failed_df.select(\"error_code\", \"error_message\"))" + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": "### Step 6b: Batch Processing (State Management Provided)\n\n`batch_process` reads from a Delta input table, writes results to an output table, and\nrecords how far it got in a metadata table. With `process_new_records_only=True`, each run\npicks up only the rows added since the last successful run, so you can schedule it without\nre-sending history.\n\n**One time steps:** Create the three Delta tables. Every future run reuses these same tables \u2014\nthe metadata table is what remembers your progress, so do not drop or recreate it between\nruns. The `setup_*` methods are safe to re-run: they return the existing table if it is\nalready there." + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": "from ttd_databricks_python.ttd_databricks import TTDEndpoint\n\ninput_table = client.setup_input_table(endpoint=TTDEndpoint.OFFLINE_CONVERSION)\noutput_table = client.setup_output_table(endpoint=TTDEndpoint.OFFLINE_CONVERSION)\nmetadata_table = client.setup_metadata_table()\n\nprint(f\"Input table: {input_table}\")\nprint(f\"Output table: {output_table}\")\nprint(f\"Metadata table: {metadata_table}\")" + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": "**Every run:**\n\n**1. Append new rows to the input table.** In production this is your upstream pipeline's job." + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": "from pyspark.sql import functions as F\n\n(\n spark.createDataFrame(rows, schema=input_schema)\n .withColumn(\"updated_at\", F.current_timestamp())\n .write.format(\"delta\").mode(\"append\").saveAsTable(input_table)\n)\n\ndisplay(spark.table(input_table))" + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": "**2. Call `batch_process`.** Re-running it picks up only rows appended since the last run." + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": "client.batch_process(\n context=context,\n input_table=input_table,\n output_table=output_table,\n metadata_table=metadata_table,\n process_new_records_only=True, # incremental; set False to reprocess every row\n batch_size=1600, # rows per API request\n)\n\ndisplay(spark.table(output_table))\ndisplay(spark.table(metadata_table))" + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": "## (Optional) Sending email addresses or phone numbers (UID2)\n\nSkip this section if you are sending device IDs or UID2s you have already resolved.\n\nPass a `uid2_config` when you create the client. It is the same call as Step 3 with one\nextra argument. Then set the `type` field inside `user_ids` to `Email`, `Phone`,\n`HashedEmail`, or `HashedPhone` on your rows. From there `push_data` and `batch_process` take them exactly like any other\nidentifier type, and nothing else about Step 6a or Step 6b changes.\n\nThe SDK resolves each identifier to a UID2 (or EUID) using your operator before the\nrequest leaves Databricks, so The Trade Desk never receives the raw email or phone\nnumber. The mapping comes back in the `uid2_resolutions` column." + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": "from datetime import datetime, timezone\n\nfrom ttd_data.uid2 import IdentityScope, UID2Config\n\nfrom ttd_databricks_python.ttd_databricks import TtdDatabricksClient\n\nTS = datetime(2026, 1, 15, 10, 11, 30, tzinfo=timezone.utc)\n\nuid2_client = TtdDatabricksClient.from_params(\n api_token=API_TOKEN,\n uid2_config=UID2Config(\n base_url=\"\",\n api_key=\"\",\n client_secret=\"\",\n identity_scope=IdentityScope.UID2, # use IdentityScope.EUID for European identities\n ),\n)\n\nuid2_data = [\n {\"tracking_tag_id\": TRACKING_TAG_ID, \"timestamp_utc\": TS,\n \"user_ids\": [{\"type\": \"Email\", \"id\": \"user@example.com\"}]},\n {\"tracking_tag_id\": TRACKING_TAG_ID, \"timestamp_utc\": TS,\n \"user_ids\": [{\"type\": \"HashedEmail\",\n \"id\": \"tMmiiTI7IaAcPpQPFQ65uMVCWH8av9jw4cwf/F5HVRQ=\"}]},\n]\n\nuid2_result_df = uid2_client.push_data(\n df=spark.createDataFrame(uid2_data, schema=input_schema),\n context=context,\n)\n\ndisplay(uid2_result_df.select(\"tracking_tag_id\", \"success\", \"error_message\", \"uid2_resolutions\"))" + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": "## Next steps\n\n- **Other use cases** \u2014 one notebook per use case:\n [First Party Data (1PD) Example Notebook](https://github.com/thetradedesk/ttd-databricks-python/blob/add-easy-start-examples-per-usecase-to-sdk-docs/example_notebook/First%20Party%20Data%20%281PD%29%20Example%20Notebook.ipynb),\n [Third Party Data (3PD) Example Notebook](https://github.com/thetradedesk/ttd-databricks-python/blob/add-easy-start-examples-per-usecase-to-sdk-docs/example_notebook/Third%20Party%20Data%20%283PD%29%20Example%20Notebook.ipynb),\n [Offline Conversion Data (CAPI) Example Notebook](https://github.com/thetradedesk/ttd-databricks-python/blob/add-easy-start-examples-per-usecase-to-sdk-docs/example_notebook/Offline%20Conversion%20Data%20%28CAPI%29%20Example%20Notebook.ipynb),\n [Deletion and Opt-Out (DSR) Example Notebook](https://github.com/thetradedesk/ttd-databricks-python/blob/add-easy-start-examples-per-usecase-to-sdk-docs/example_notebook/Deletion%20and%20Opt-Out%20%28DSR%29%20Example%20Notebook.ipynb).\n- **Full reference** \u2014 [README](https://github.com/thetradedesk/ttd-databricks-python/blob/add-easy-start-examples-per-usecase-to-sdk-docs/README.md) covers authentication, error handling,\n UID2 support, custom HTTP clients, and server URL overrides.\n- **Server URLs** \u2014 you do not need to configure one. Each endpoint already targets its\n own default server, and it can be overridden with a preferred server if you need one." + } + ], + "metadata": { + "kernelspec": { + "display_name": "Python 3", + "language": "python", + "name": "python3" + }, + "language_info": { + "name": "python", + "version": "3.10.0" + } + }, + "nbformat": 4, + "nbformat_minor": 4 +} diff --git a/example_notebook/TTD Connector Data SDK Example Notebook.ipynb b/example_notebook/TTD Connector Data SDK Example Notebook.ipynb deleted file mode 100644 index e507827..0000000 --- a/example_notebook/TTD Connector Data SDK Example Notebook.ipynb +++ /dev/null @@ -1,333 +0,0 @@ -{ - "cells": [ - { - "cell_type": "markdown", - "metadata": {}, - "source": [ - "# TTD Databricks SDK - Example Notebook\n", - "\n", - "This notebook demonstrates how to use the `ttd-databricks` SDK to submit audience data\n", - "to The Trade Desk's Data API from a Databricks environment.\n", - "\n", - "## Prerequisites\n", - "- Package installed: `%pip install ttd-databricks`\n", - "- A valid TTD API token (TTD-Auth)\n", - "- A valid TTD Data Provider ID\n", - "- A valid TTD Advertiser ID (for advertiser endpoint)" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "%pip install ttd-databricks\n", - "\n", - "# Recommended to restart kerner to use updated packages\n", - "dbutils.library.restartPython()" - ] - }, - { - "cell_type": "markdown", - "metadata": {}, - "source": [ - "## Step 1: Configure credentials\n", - "\n", - "In production, retrieve secrets from Databricks Secrets:\n", - "```python\n", - "api_token = dbutils.secrets.get(scope=\"ttd\", key=\"api-token\")\n", - "```\n", - "\n", - "> **Note:** The SDK does not support `TtdSignature` based authentication. Refer to [OpenTTD](https://open.thetradedesk.com/advertiser/docsApp/Foundations/resources/doc/PlatformAuthentication) for instructions on how to create your `TTD-Auth` API token (select `Data API` as the `Application`)." - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "# Replace with your actual values\n", - "API_TOKEN = \"\"\n", - "DATA_PROVIDER_ID = \"\"\n", - "ADVERTISER_ID = \"\"" - ] - }, - { - "cell_type": "markdown", - "metadata": {}, - "source": [ - "## Step 2: Create the client\n", - "\n", - "Use `from_params()` to create the client from your credentials.\n", - "The SparkSession is auto-detected from the Databricks runtime." - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "from ttd_databricks_python.ttd_databricks import (\n", - " TtdDatabricksClient,\n", - " AdvertiserContext,\n", - " TTDEndpoint,\n", - " get_ttd_input_schema,\n", - " TTDSchemaValidationError,\n", - ")\n", - "\n", - "client = TtdDatabricksClient.from_params(api_token=API_TOKEN)\n", - "print(\"Client ready.\")" - ] - }, - { - "cell_type": "markdown", - "metadata": {}, - "source": [ - "## Step 3: Create a context\n", - "\n", - "The context holds endpoint-specific config: which advertiser and data provider\n", - "the data belongs to." - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "context = AdvertiserContext(\n", - " data_provider_id=DATA_PROVIDER_ID,\n", - " advertiser_id=ADVERTISER_ID,\n", - ")\n", - "print(f\"Context: {context}\")" - ] - }, - { - "cell_type": "markdown", - "metadata": {}, - "source": [ - "## Step 4: Inspect the required input schema\n", - "\n", - "Use `get_ttd_input_schema()` to see which columns your DataFrame must contain.\n", - "\n", - "For more information on the meaning of particular fields and supported data types supported per endpoint refer to the following table:\n", - "\n", - "| Endpoint | Context | Data API | OpenTTD API Documentation |\n", - "|---|---|---|---|\n", - "| Advertiser | `AdvertiserContext` | `POST /data/advertiser` | [OpenTTD Documentation](https://open.thetradedesk.com/advertiser/docsApp/GuidesAdvertiser/data/doc/post-data-advertiser-firstparty)
[OpenTTD Documentation (External Provider)](https://open.thetradedesk.com/provider/docsApp/GuidesProvider/audience/doc/post-data-advertiser-external) |\n", - "| Third Party | `ThirdPartyContext` | `POST /data/thirdparty` | [OpenTTD Documentation](https://open.thetradedesk.com/provider/docsApp/GuidesProvider/audience/doc/post-data-thirdparty) |\n", - "| Offline Conversion | `OfflineConversionContext` | `POST /providerapi/offlineconversion` | [OpenTTD Documentation](https://open.thetradedesk.com/advertiser/docsApp/GuidesAdvertiser/data/doc/post-providerapi-offlineconversion) |\n", - "| Deletion / Opt-Out — Advertiser | `DeletionOptOutAdvertiserContext` | `POST /data/deletion-optout/advertiser` | [OpenTTD Documentation](https://open.thetradedesk.com/advertiser/docsApp/GuidesAdvertiser/data/doc/post-data-deletion-optout-advertiser)
[OpenTTD Documentation (External Provider)](https://open.thetradedesk.com/provider/docsApp/GuidesProvider/audience/doc/post-data-deletion-optout-advertiser-external) |\n", - "| Deletion / Opt-Out — Third Party | `DeletionOptOutThirdPartyContext` | `POST /data/deletion-optout/thirdparty` | [OpenTTD Documentation](https://open.thetradedesk.com/provider/docsApp/GuidesProvider/audience/doc/post-data-deletion-optout-thirdparty) |\n", - "| Deletion / Opt-Out — Merchant | `DeletionOptOutMerchantContext` | `POST /data/deletion-optout/merchant` | [OpenTTD Documentation](https://open.thetradedesk.com/provider/docsApp/GuidesProvider/retail/doc/post-data-deletion-optout-merchant) |" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "input_schema = get_ttd_input_schema(TTDEndpoint.ADVERTISER)\n", - "print(\"Required input schema:\")\n", - "\n", - "for field in input_schema.fields:\n", - " print(f\" {field.name}: {field.dataType.simpleString()} (nullable={field.nullable})\")" - ] - }, - { - "cell_type": "markdown", - "metadata": {}, - "source": [ - "## Step 5: Prepare your input DataFrame\n", - "\n", - "Your DataFrame must contain at minimum the mandatory columns.\n", - "Extra columns are allowed and will be preserved in the output." - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "# Example: create a small sample DataFrame\n", - "# In practice, read from a Delta table or other data source\n", - "sample_data = [\n", - " {\"id_type\": \"TDID\", \"id_value\": \"a3f1c2d4-8e7b-4f6a-9c0d-1b2e3f4a5b6c\", \"segment_name\": \"segment_1\"},\n", - " {\"id_type\": \"DAID\", \"id_value\": \"7d9e0f1a-2b3c-4d5e-6f7a-8b9c0d1e2f3a\", \"segment_name\": \"segment_2\"},\n", - " # intentionally incorrect format for ramp_id to showcase error enrties in output\n", - " {\"id_type\": \"RampID\", \"id_value\": \"c4d5e6f7-a8b9-4c0d-1e2f-3a4b5c6d7e8f\", \"segment_name\": \"segment_3\"},\n", - " {\"id_type\": \"TDID\", \"id_value\": \"1f2a3b4c-5d6e-4f7a-8b9c-0d1e2f3a4b5c\", \"segment_name\": \"segment_4\"},\n", - " {\"id_type\": \"DAID\", \"id_value\": \"9b0c1d2e-3f4a-4b5c-6d7e-8f9a0b1c2d3e\", \"segment_name\": \"segment_5\"},\n", - "]\n", - "\n", - "input_df = spark.createDataFrame(sample_data)\n", - "display(input_df)" - ] - }, - { - "cell_type": "markdown", - "metadata": {}, - "source": [ - "## Step 6: Submit data via push_data (ad hoc mode)\n", - "\n", - "`push_data` will:\n", - "1. Validate your DataFrame has the required columns\n", - "2. Send data in batches (default: 1600 rows per request)\n", - "3. Return a new DataFrame with all original columns plus status columns:\n", - " `success`, `error_code`, `error_message`, `processed_timestamp`" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "try:\n", - " result_df = client.push_data(\n", - " df=input_df,\n", - " context=context,\n", - " batch_size=1600, # Number of rows batched together in a single request to The Trade Desk\n", - " )\n", - " display(result_df)\n", - "except TTDSchemaValidationError as e:\n", - " print(f\"Schema validation failed: {e}\")" - ] - }, - { - "cell_type": "markdown", - "metadata": {}, - "source": [ - "## Step 7: Inspect results" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "from pyspark.sql.functions import col\n", - "\n", - "total = result_df.count()\n", - "succeeded = result_df.filter(col(\"success\") == True).count()\n", - "failed = total - succeeded\n", - "\n", - "print(f\"Total rows: {total}\")\n", - "print(f\"Succeeded: {succeeded}\")\n", - "print(f\"Failed: {failed}\")\n", - "\n", - "if failed > 0:\n", - " print(\"\\nFailed rows:\")\n", - " display(result_df.filter(col(\"success\") == False))" - ] - }, - { - "cell_type": "markdown", - "metadata": {}, - "source": [ - "## Optional: Table setup for batch processing mode\n", - "\n", - "Use the table setup utilities to create Delta tables with the correct schema,\n", - "then use `batch_process()` for ongoing incremental processing." - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "# Creates three managed Delta tables in the active catalog/database.\n", - "# Default names: ttd_advertiser_input, ttd_advertiser_output, ttd_metadata\n", - "# Pass table_name= and location= to use custom names or external storage.\n", - "input_table = client.setup_input_table(endpoint=TTDEndpoint.ADVERTISER)\n", - "output_table = client.setup_output_table(endpoint=TTDEndpoint.ADVERTISER)\n", - "metadata_table = client.setup_metadata_table(table_name=\"ttd_advertiser_metadata\")\n", - "\n", - "print(f\"Input table: {input_table}\")\n", - "print(f\"Output table: {output_table}\")\n", - "print(f\"Metadata table: {metadata_table}\")" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "from pyspark.sql import functions as F\n", - "\n", - "# This cell simulates your upstream pipeline writing records to the input table\n", - "# The user is responsible to write into the input table, the SDK only performs reads from the table\n", - "\n", - "\n", - "# updated_at is required for incremental processing: batch_process uses it\n", - "# to filter rows added since the last run when process_new_records_only=True\n", - "# The user is responsible to set the updated_at value for entries in the input table\n", - "\n", - "(\n", - " spark.createDataFrame(sample_data)\n", - " .withColumn(\"updated_at\", F.current_timestamp())\n", - " .write.format(\"delta\")\n", - " .mode(\"append\")\n", - " .saveAsTable(input_table)\n", - ")\n", - "\n", - "display(spark.table(input_table))" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "# Run batch processing (reads from input_table, writes to output_table)\n", - "\n", - "# process_new_records_only=True filters to rows where updated_at > last run date\n", - "# On the first run, metadata_table is empty so all rows are processed\n", - "client.batch_process(\n", - " context=context,\n", - " input_table=input_table,\n", - " output_table=output_table,\n", - " metadata_table=metadata_table,\n", - " process_new_records_only=True, # Processes rows updated after last run; processes all rows on first run\n", - " batch_size=1600, # Number of rows grouped together in a single request to The Trade Desk\n", - " parallelism=16, # Number of paralellel workers processing the entries from the input table\n", - ")" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "# Display the output table\n", - "display(spark.table(output_table))\n", - "\n", - "# Display the metadata table\n", - "display(spark.table(metadata_table))" - ] - } - ], - "metadata": { - "kernelspec": { - "display_name": "Python 3", - "language": "python", - "name": "python3" - }, - "language_info": { - "name": "python", - "version": "3.10.0" - } - }, - "nbformat": 4, - "nbformat_minor": 4 -} \ No newline at end of file diff --git a/example_notebook/Third Party Data (3PD) Example Notebook.ipynb b/example_notebook/Third Party Data (3PD) Example Notebook.ipynb new file mode 100644 index 0000000..8da4644 --- /dev/null +++ b/example_notebook/Third Party Data (3PD) Example Notebook.ipynb @@ -0,0 +1,164 @@ +{ + "cells": [ + { + "cell_type": "markdown", + "metadata": {}, + "source": "# Third Party Data (3PD) Example Notebook\n\nSend audience segments you own as a data provider to The Trade Desk for monetization.\n\n## Is this the right notebook for you?\n\n| | |\n|---|---|\n| **Who Should Use This?** | Third-party data providers and commerce partners |\n| **What Does the Notebook Do?** | Adds identities to a third-party audience segment available in the TTD marketplace |\n| **What Data Does It Send?** | One row per identity per segment |\n| **Destination Trade Desk Endpoint** | `POST /data/thirdparty` (via the `ThirdPartyContext` class of the ttd-databricks SDK) |\n| **Relevant OpenTTD API Documentation** | [Third-party data](https://open.thetradedesk.com/provider/docsApp/GuidesProvider/audience/doc/post-data-thirdparty) |\n\nUploading data on behalf of a single advertiser instead? That is first-party data, so use\n[First Party Data (1PD) Example Notebook](https://github.com/thetradedesk/ttd-databricks-python/blob/add-easy-start-examples-per-usecase-to-sdk-docs/example_notebook/First%20Party%20Data%20%281PD%29%20Example%20Notebook.ipynb).\n\n## What you need\n\n- [ ] A TTD Platform API token \u2014 see [Create an API token](https://open.thetradedesk.com/advertiser/docsApp/GuidesAdvertiser/data/doc/DataApiCallsAdvertiser#ui-method-create)\n- [ ] Your data provider ID\n- [ ] (Optional) A UID2 operator base URL, API key and client secret \u2014 to send raw email addresses or phone numbers" + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": "## Step 1: Install the SDK" + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": "%pip install ttd-databricks\n\ndbutils.library.restartPython()" + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": "## Step 2: Configure credentials\n\nFor a first run you can paste values inline. In production, read them from Databricks Secrets:\n\n```python\nAPI_TOKEN = dbutils.secrets.get(scope=\"ttd\", key=\"api-token\")\nDATA_PROVIDER_ID = dbutils.secrets.get(scope=\"ttd\", key=\"data-provider-id\")\n```\n\n> **Note:** Authenticate with a Platform API token, sent as the `TTD-Auth` header. Generate one\n> in the OpenTTD Access Management app \u2014 see [Create an API token](https://open.thetradedesk.com/advertiser/docsApp/GuidesAdvertiser/data/doc/DataApiCallsAdvertiser#ui-method-create). Secret keys and\n> `TtdSignature` headers are a legacy method and are not supported by this SDK." + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": "API_TOKEN = \"\"\nDATA_PROVIDER_ID = \"\"" + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": "## Step 3: Create the client and context" + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": "from ttd_databricks_python.ttd_databricks import (\n ThirdPartyContext,\n TTDEndpoint,\n TtdDatabricksClient,\n get_ttd_input_schema,\n)\n\n# Two ways to create the client \u2014 this notebook uses (ii) everywhere below.\n# i) Dependency injection \u2014 you build the DataClient and pass it in:\n# from ttd_data import DataClient\n# client = TtdDatabricksClient(data_api_client=DataClient(), api_token=API_TOKEN)\n# ii) Factory \u2014 from_params builds the DataClient for you:\nclient = TtdDatabricksClient.from_params(api_token=API_TOKEN)\n\ncontext = ThirdPartyContext(\n data_provider_id=DATA_PROVIDER_ID,\n # Set True only if id_value already holds a hashed identifier.\n is_user_id_already_hashed=False,\n)\n\nprint(f\"Context: {context}\")" + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": "## Step 4: Inspect the required input schema\n\nMandatory columns: `id_type`, `id_value`, `segment_name`\n\nOptional columns: `cookie_mapping_partner_id`, `timestamp_utc`, `ttl_in_minutes`" + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": "from ttd_databricks_python.ttd_databricks import TTDEndpoint, get_ttd_input_schema\nfrom ttd_databricks_python.ttd_databricks.schemas import get_required_column_names\n\ninput_schema = get_ttd_input_schema(TTDEndpoint.THIRD_PARTY)\n\nprint(\"Mandatory columns:\", get_required_column_names(TTDEndpoint.THIRD_PARTY))\nprint(\"\\nFull input schema:\")\nfor field in input_schema.fields:\n print(f\" {field.name}: {field.dataType.simpleString()} (nullable={field.nullable})\")" + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": "## Step 5: Prepare your input DataFrame\n\nOne row per identity per segment. `segment_name` is the data element the identity joins.\n\n> **Tip:** Start with a handful of rows. Confirming the end-to-end flow on a small sample is much easier to troubleshoot than a full load." + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": "rows = [\n {\"id_type\": \"TDID\", \"id_value\": \"123e4567-e89b-12d3-a456-426652340000\",\n \"segment_name\": \"1210\", \"ttl_in_minutes\": 43200},\n {\"id_type\": \"DAID\", \"id_value\": \"a9342d1f-69f1-4bf8-bc2b-1f20eb451f21\",\n \"segment_name\": \"1150\", \"ttl_in_minutes\": 43200},\n {\"id_type\": \"UID2\", \"id_value\": \"48MjlfIUZpOKNAm9nod7/jCLAXUYsnE1tpVHQSDS0uo=\",\n \"segment_name\": \"1630\", \"ttl_in_minutes\": 43200},\n {\"id_type\": \"ID5\", \"id_value\": \"ID5-c62drGF0EC6wsCZVFDbTbZwi33eB0uZTIC8FxJpzsQ\",\n \"segment_name\": \"1800\", \"ttl_in_minutes\": 43200},\n {\"id_type\": \"FirstId\", \"id_value\": \"8934d279bba4c7d652a02f624dc334e3\",\n \"segment_name\": \"1810\", \"ttl_in_minutes\": 43200},\n]\n\ninput_df = spark.createDataFrame(rows, schema=input_schema)\ndisplay(input_df)" + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": "## Step 6: Sending Data\n\nThere are two ways to send your data. Pick one \u2014 you do not need both.\n\n| | Ad hoc | Batch processing |\n|---|---|---|\n| **State management** | None. Every call sends every row you give it. | Provided. A metadata table records progress, so each run sends only rows added since the last one. |\n| **Input** | A DataFrame you build in the notebook | A Delta input table |\n| **Output** | Returned inline as a DataFrame | Written to a Delta output table |\n| **Best for** | One-off loads and first tests | Recurring pipelines |" + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": "### Step 6a: Ad Hoc Usage (No State Management Provided)\n\n`push_data` sends your DataFrame straight to the Data API and returns the input columns\nenriched with per-row status. The SDK keeps no record of what it has already sent, so\nre-running this cell sends every row again.\n\nIt does not raise on API or row-level failures \u2014 every outcome is reported inline." + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": "result_df = client.push_data(df=input_df, context=context, batch_size=1600)\n\ndisplay(result_df)" + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": "`push_data` adds these columns to your input:\n\n| Column | Meaning |\n|---|---|\n| `success` | `True` if the row was accepted |\n| `error_code` | Failure category, `null` on success |\n| `error_message` | Human-readable reason, `null` on success |\n| `processed_timestamp` | When the row was submitted |\n| `uid2_resolutions` | Raw identifier \u2192 UID2 mapping, empty unless `uid2_config` was set |" + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": "from pyspark.sql.functions import col\n\ntotal = result_df.count()\nsucceeded = result_df.filter(col(\"success\")).count()\n\nprint(f\"Total: {total} | Succeeded: {succeeded} | Failed: {total - succeeded}\")\n\nfailed_df = result_df.filter(~col(\"success\"))\nif failed_df.count():\n display(failed_df.select(\"error_code\", \"error_message\"))" + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": "### Step 6b: Batch Processing (State Management Provided)\n\n`batch_process` reads from a Delta input table, writes results to an output table, and\nrecords how far it got in a metadata table. With `process_new_records_only=True`, each run\npicks up only the rows added since the last successful run, so you can schedule it without\nre-sending history.\n\n**One time steps:** Create the three Delta tables. Every future run reuses these same tables \u2014\nthe metadata table is what remembers your progress, so do not drop or recreate it between\nruns. The `setup_*` methods are safe to re-run: they return the existing table if it is\nalready there." + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": "from ttd_databricks_python.ttd_databricks import TTDEndpoint\n\ninput_table = client.setup_input_table(endpoint=TTDEndpoint.THIRD_PARTY)\noutput_table = client.setup_output_table(endpoint=TTDEndpoint.THIRD_PARTY)\nmetadata_table = client.setup_metadata_table()\n\nprint(f\"Input table: {input_table}\")\nprint(f\"Output table: {output_table}\")\nprint(f\"Metadata table: {metadata_table}\")" + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": "**Every run:**\n\n**1. Append new rows to the input table.** In production this is your upstream pipeline's job." + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": "from pyspark.sql import functions as F\n\n(\n spark.createDataFrame(rows, schema=input_schema)\n .withColumn(\"updated_at\", F.current_timestamp())\n .write.format(\"delta\").mode(\"append\").saveAsTable(input_table)\n)\n\ndisplay(spark.table(input_table))" + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": "**2. Call `batch_process`.** Re-running it picks up only rows appended since the last run." + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": "client.batch_process(\n context=context,\n input_table=input_table,\n output_table=output_table,\n metadata_table=metadata_table,\n process_new_records_only=True, # incremental; set False to reprocess every row\n batch_size=1600, # rows per API request\n)\n\ndisplay(spark.table(output_table))\ndisplay(spark.table(metadata_table))" + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": "## (Optional) Sending email addresses or phone numbers (UID2)\n\nSkip this section if you are sending device IDs or UID2s you have already resolved.\n\nPass a `uid2_config` when you create the client. It is the same call as Step 3 with one\nextra argument. Then set `id_type` to `Email`, `Phone`,\n`HashedEmail`, or `HashedPhone` on your rows. From there `push_data` and `batch_process` take them exactly like any other\nidentifier type, and nothing else about Step 6a or Step 6b changes.\n\nThe SDK resolves each identifier to a UID2 (or EUID) using your operator before the\nrequest leaves Databricks, so The Trade Desk never receives the raw email or phone\nnumber. The mapping comes back in the `uid2_resolutions` column." + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": "from ttd_data.uid2 import IdentityScope, UID2Config\n\nfrom ttd_databricks_python.ttd_databricks import TtdDatabricksClient\n\nuid2_client = TtdDatabricksClient.from_params(\n api_token=API_TOKEN,\n uid2_config=UID2Config(\n base_url=\"\",\n api_key=\"\",\n client_secret=\"\",\n identity_scope=IdentityScope.UID2, # use IdentityScope.EUID for European identities\n ),\n)\n\nuid2_data = [\n {\"id_type\": \"Email\", \"id_value\": \"user@example.com\",\n \"segment_name\": \"1210\", \"ttl_in_minutes\": 43200},\n {\"id_type\": \"HashedEmail\", \"id_value\": \"tMmiiTI7IaAcPpQPFQ65uMVCWH8av9jw4cwf/F5HVRQ=\",\n \"segment_name\": \"1210\", \"ttl_in_minutes\": 43200},\n]\n\nuid2_result_df = uid2_client.push_data(\n df=spark.createDataFrame(uid2_data, schema=input_schema),\n context=context,\n)\n\ndisplay(uid2_result_df.select(\"id_type\", \"success\", \"error_message\", \"uid2_resolutions\"))" + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": "## Next steps\n\n- **Other use cases** \u2014 one notebook per use case:\n [First Party Data (1PD) Example Notebook](https://github.com/thetradedesk/ttd-databricks-python/blob/add-easy-start-examples-per-usecase-to-sdk-docs/example_notebook/First%20Party%20Data%20%281PD%29%20Example%20Notebook.ipynb),\n [Third Party Data (3PD) Example Notebook](https://github.com/thetradedesk/ttd-databricks-python/blob/add-easy-start-examples-per-usecase-to-sdk-docs/example_notebook/Third%20Party%20Data%20%283PD%29%20Example%20Notebook.ipynb),\n [Offline Conversion Data (CAPI) Example Notebook](https://github.com/thetradedesk/ttd-databricks-python/blob/add-easy-start-examples-per-usecase-to-sdk-docs/example_notebook/Offline%20Conversion%20Data%20%28CAPI%29%20Example%20Notebook.ipynb),\n [Deletion and Opt-Out (DSR) Example Notebook](https://github.com/thetradedesk/ttd-databricks-python/blob/add-easy-start-examples-per-usecase-to-sdk-docs/example_notebook/Deletion%20and%20Opt-Out%20%28DSR%29%20Example%20Notebook.ipynb).\n- **Full reference** \u2014 [README](https://github.com/thetradedesk/ttd-databricks-python/blob/add-easy-start-examples-per-usecase-to-sdk-docs/README.md) covers authentication, error handling,\n UID2 support, custom HTTP clients, and server URL overrides.\n- **Server URLs** \u2014 you do not need to configure one. Each endpoint already targets its\n own default server, and it can be overridden with a preferred server if you need one." + } + ], + "metadata": { + "kernelspec": { + "display_name": "Python 3", + "language": "python", + "name": "python3" + }, + "language_info": { + "name": "python", + "version": "3.10.0" + } + }, + "nbformat": 4, + "nbformat_minor": 4 +}