Integration Guides

Appendix Z: Integration Guides

Apache Spark Integration

Write Spark streaming and batch DataFrames directly to DEML ingestion endpoints for unified telemetry and ML feature pipelines.

Prerequisites

• Apache Spark 3.4+ (Structured Streaming or batch)
• Network egress to https://backend.deml.app
• DEML API key stored in your cluster secrets manager

Batch Write Pattern

Transform your DataFrame and POST batches via a mapPartitions sink:

import json
import requests
from pyspark.sql import SparkSession

API_KEY = "YOUR_API_KEY"  # pragma: allowlist secret
INGEST_URL = "https://backend.deml.app/api/v1/ingest"

spark = SparkSession.builder.appName("deml-ingest").getOrCreate()
df = spark.read.parquet("s3://datalake/events/")

def send_partition(rows):
    records = [row.asDict() for row in rows]
    if not records:
        return
    requests.post(
        INGEST_URL,
        headers={"Authorization": f"Bearer {API_KEY}"},
        json={"source": "spark", "records": records},
        timeout=60,
    ).raise_for_status()

df.foreachPartition(send_partition)

Structured Streaming

Stream micro-batches to DEML as they arrive:

from pyspark.sql.functions import col, struct, to_json

stream = (
    spark.readStream.format("kafka")
    .option("kafka.bootstrap.servers", "broker:9092")
    .option("subscribe", "telemetry")
    .load()
)

payload = stream.select(
    to_json(struct(col("value").alias("payload"))).alias("record")
)

def write_batch(batch_df, batch_id):
    rows = [row.record for row in batch_df.collect()]
    if rows:
        requests.post(
            INGEST_URL,
            headers={"Authorization": f"Bearer {API_KEY}"},
            json={"source": "spark-stream", "batch_id": batch_id, "records": rows},
            timeout=60,
        ).raise_for_status()

query = payload.writeStream.foreachBatch(write_batch).start()

Scala Alternative

df.writeStream
  .format("org.apache.spark.sql.kafka010.KafkaSourceProvider")
  .option("checkpointLocation", "/checkpoints/deml")
  .foreachBatch { (batchDF: DataFrame, batchId: Long) =>
    val records = batchDF.collect().map(_.getAs[String]("payload"))
    // POST records to https://backend.deml.app/api/v1/ingest
  }
  .start()

Planned Native Spark Connector

A first-class deml format will simplify writes:

df.writeStream
  .format("deml")
  .option("api_key", sys.env("DEML_API_KEY"))
  .option("endpoint", "https://backend.deml.app/api/v1/ingest")
  .start()

Integration Health Check

curl https://backend.deml.app/api/v1/integrations/apache-spark \
  -H "Authorization: Bearer YOUR_API_KEY"

Databricks Integration

Connect Databricks notebooks and jobs to DEML for secure telemetry ingest, model inference, and cross-platform analytics.

Prerequisites

• Databricks Runtime 13.3+ (Python or Scala)
• DEML API key
• Outbound HTTPS to backend.deml.app

Store Credentials in Databricks Secrets

Never hardcode API keys in notebooks. Use a Secret Scope:

databricks secrets create-scope --scope deml
databricks secrets put --scope deml --key api-key --string-value YOUR_API_KEY

In a notebook:

api_key = dbutils.secrets.get(scope="deml", key="api-key")  # pragma: allowlist secret

Ingest from a Notebook

import requests

INGEST_URL = "https://backend.deml.app/api/v1/ingest"
api_key = dbutils.secrets.get(scope="deml", key="api-key")

df = spark.table("analytics.telemetry_events")
records = [row.asDict() for row in df.limit(1000).collect()]

response = requests.post(
    INGEST_URL,
    headers={"Authorization": f"Bearer {api_key}"},
    json={"source": "databricks", "records": records},
    timeout=60,
)
response.raise_for_status()
print(f"Ingested {len(records)} records")

Scheduled Job Pattern

1. Create a Databricks Job with a Python task.
2. Mount the deml secret scope on the cluster.
3. Run on a schedule (e.g., every 5 minutes) to push aggregated features.

# Databricks job: push hourly rollups to DEML
rollup = spark.sql("""
  SELECT tenant_id, AVG(latency_ms) AS avg_latency, COUNT(*) AS requests
  FROM delta.`/mnt/telemetry/raw`
  WHERE event_time > current_timestamp() - INTERVAL 1 HOUR
  GROUP BY tenant_id
""")

records = rollup.collect()
requests.post(
    INGEST_URL,
    headers={"Authorization": f"Bearer {api_key}"},
    json={"source": "databricks-job", "records": [r.asDict() for r in records]},
).raise_for_status()

Real-time Inference from Databricks

PREDICT_URL = "https://backend.deml.app/api/v1/predict"

def predict_row(features: list[float]) -> float:
    result = requests.post(
        PREDICT_URL,
        headers={"Authorization": f"Bearer {api_key}"},
        json={"model_version": "v2", "inputs": features},
        timeout=10,
    )
    result.raise_for_status()
    return result.json()["outputs"][0]

# Apply to a Spark UDF or driver-side batch calls
scores = [predict_row(row.features) for row in df.limit(100).collect()]

Unity Catalog & Multi-Tenancy

Map Databricks workspace catalogs to DEML tenant UUIDs in your job metadata so analytics remain isolated per customer.

Integration Health Check

curl https://backend.deml.app/api/v1/integrations/databricks \
  -H "Authorization: Bearer YOUR_API_KEY"

Kubernetes Integration

Integrating the DEML platform into your Kubernetes cluster lets microservices stream telemetry and request predictions through our API Gateway without leaving your cluster boundary.

Architecture Options

Pattern	Best for	Latency	Ops overhead
Sidecar proxy	Per-pod inference + ingest	Lowest	Medium
Cluster gateway	Shared ingress for many services	Low	Low
CRD / Operator (roadmap)	Declarative pipeline provisioning	Low	Lowest at scale

Sidecar Proxy Pattern (Recommended)

Deploy a lightweight sidecar alongside your application pods. The sidecar injects your API key, handles rate-limit backoff, and forwards traffic to /api/v1/predict and /api/v1/ingest.

1. Store your API key in a Secret

apiVersion: v1
kind: Secret
metadata:
  name: deml-platform-credentials
  namespace: production
type: Opaque
stringData:
  api-key: YOUR_API_KEY

2. Configure the sidecar in your Pod spec

apiVersion: v1
kind: Pod
metadata:
  name: ml-inference-service
  labels:
    app: ml-inference
spec:
  containers:
    - name: app
      image: your-registry/inference-app:latest
      env:
        - name: DEML_GATEWAY_URL
          value: "http://127.0.0.1:8080"
    - name: deml-sidecar
      image: ghcr.io/deml/sidecar-proxy:latest
      ports:
        - containerPort: 8080
      env:
        - name: DEML_UPSTREAM_URL
          value: "https://backend.deml.app/api/v1"
        - name: DEML_API_KEY
          valueFrom:
            secretKeyRef:
              name: deml-platform-credentials
              key: api-key

Your application calls http://127.0.0.1:8080/predict locally; the sidecar adds authentication and forwards to DEML.

3. Verify connectivity

kubectl exec -it ml-inference-service -c app -- \
  curl -s http://127.0.0.1:8080/health

Check integration status from your cluster (optional health endpoint):

curl https://backend.deml.app/api/v1/integrations/kubernetes \
  -H "Authorization: Bearer YOUR_API_KEY"

Cluster Gateway Pattern

For shared access across namespaces, expose a single internal Service that proxies to DEML:

apiVersion: v1
kind: Service
metadata:
  name: deml-gateway
  namespace: platform
spec:
  selector:
    app: deml-gateway
  ports:
    - port: 443
      targetPort: 8443

Point workloads at https://deml-gateway.platform.svc.cluster.local and mount the API key via External Secrets or GCP Secret Manager.

Telemetry Ingest from Kubernetes

Stream pod metrics, request logs, or custom events to /api/v1/ingest:

curl -X POST https://backend.deml.app/api/v1/ingest \
  -H "Authorization: Bearer YOUR_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "source": "kubernetes",
    "cluster_id": "prod-us-east-1",
    "records": [
      {"pod": "inference-7f8b", "latency_ms": 42, "status": 200}
    ]
  }'

Events flow through Redpanda → telemetry workers → your analytics dashboard in real time.

Roadmap: Kubernetes Operator

We are developing a native MLPlatform CRD so you can declare inference routes and ingestion pipelines in Git:

apiVersion: deml.app/v1
kind: InferenceRoute
metadata:
  name: sla-model
spec:
  modelVersion: v2
  replicas: 3
  tenantId: YOUR_TENANT_UUID

Subscribe to release notes for operator availability.

PyTorch Integration

Use DEML as a remote data source and inference backend from PyTorch training scripts, DataLoaders, and deployment pipelines.

Prerequisites

• Python 3.11+
• PyTorch 2.x
• A DEML API key (Settings → API Keys)

Custom DataLoader (Available Today)

from __future__ import annotations

import requests
import torch
from torch.utils.data import Dataset, DataLoader

API_KEY = "YOUR_API_KEY"  # pragma: allowlist secret
INGEST_URL = "https://backend.deml.app/api/v1/ingest"
PREDICT_URL = "https://backend.deml.app/api/v1/predict"


class DemlRemoteDataset(Dataset):
    def __init__(self, page_size: int = 64) -> None:
        self.page_size = page_size
        self._cache: list[tuple[torch.Tensor, torch.Tensor]] = []
        self._index = 0
        self._refresh()

    def _refresh(self) -> None:
        response = requests.post(
            INGEST_URL,
            headers={"Authorization": f"Bearer {API_KEY}"},
            json={"batch_size": self.page_size, "format": "pytorch"},
            timeout=30,
        )
        response.raise_for_status()
        records = response.json()["records"]
        self._cache = [
            (torch.tensor(r["features"], dtype=torch.float32), torch.tensor(r["label"]))
            for r in records
        ]
        self._index = 0

    def __len__(self) -> int:
        return len(self._cache)

    def __getitem__(self, idx: int) -> tuple[torch.Tensor, torch.Tensor]:
        if idx >= len(self._cache):
            self._refresh()
        return self._cache[idx % len(self._cache)]


loader = DataLoader(DemlRemoteDataset(page_size=64), batch_size=32, shuffle=True)

for features, labels in loader:
    outputs = model(features)
    loss = criterion(outputs, labels)
    loss.backward()

Remote Inference

import requests
import torch

payload = {"model_version": "v2", "inputs": [0.5, 0.2, 0.9]}
response = requests.post(
    PREDICT_URL,
    headers={"Authorization": f"Bearer {API_KEY}"},
    json=payload,
    timeout=5,
)
outputs = torch.tensor(response.json()["outputs"])

DEML hosts tenant-namespaced PyTorch state_dict checkpoints on Hugging Face — no pickle, security-first.

Planned SDK

from deml.pytorch import PlatformDataLoader

loader = PlatformDataLoader(
    api_key="YOUR_API_KEY",  # pragma: allowlist secret
    batch_size=64,
    shuffle=True,
)

for batch in loader:
    predictions = model(batch)

Integration Health Check

curl https://backend.deml.app/api/v1/integrations/pytorch \
  -H "Authorization: Bearer YOUR_API_KEY"

AWS Redshift Integration

Connect Amazon Redshift warehouses to DEML for scheduled analytics exports, feature-store rollups, and ML training pipelines. Redshift UNLOAD and COPY patterns push curated datasets into /api/v1/ingest while keeping credentials in AWS Secrets Manager or IAM roles.

Prerequisites

• Amazon Redshift cluster or Redshift Serverless workgroup
• Network egress to https://backend.deml.app (or VPC endpoint + NAT)
• DEML API key stored in AWS Secrets Manager
• Optional: S3 staging bucket for UNLOAD/COPY workflows

Architecture Options

Pattern	Best for	Latency	Ops overhead
Scheduled UNLOAD → S3	Nightly feature rollups, batch ingest	Minutes	Low
Lambda + UNLOAD	Event-driven exports after ETL	Seconds	Medium
Redshift Data API	Serverless queries without JDBC	Variable	Low
Spectrum + Spark sink	Lakehouse federated queries	Minutes	Medium

Scheduled UNLOAD to DEML Ingest

Export aggregated metrics from Redshift to S3, then POST batches to DEML:

UNLOAD (
  'SELECT tenant_id, metric_name, metric_value, recorded_at
   FROM analytics.daily_rollups
   WHERE recorded_at >= CURRENT_DATE - 1'
)
TO 's3://your-bucket/deml-export/'
IAM_ROLE 'arn:aws:iam::123456789012:role/RedshiftUnloadRole'
FORMAT AS PARQUET
ALLOWOVERWRITE;

Python job (Lambda, ECS, or Databricks) reads Parquet and ingests:

import json
import boto3
import requests

API_KEY = "YOUR_API_KEY"  # pragma: allowlist secret
INGEST_URL = "https://backend.deml.app/api/v1/ingest"
s3 = boto3.client("s3")

def ingest_parquet_object(bucket: str, key: str) -> None:
    obj = s3.get_object(Bucket=bucket, Key=key)
    # Parse Parquet with pyarrow/polars in production
    records = [{"source": "redshift", "payload": obj["Body"].read().decode("utf-8", errors="ignore")}]
    requests.post(
        INGEST_URL,
        headers={"Authorization": f"Bearer {API_KEY}"},
        json={"batch_id": key, "source": "aws-redshift", "records": records},
        timeout=120,
    ).raise_for_status()

Redshift Data API (Serverless)

Query without persistent JDBC connections and stream rows to DEML:

import boto3
import requests

redshift = boto3.client("redshift-data")
API_KEY = "YOUR_API_KEY"  # pragma: allowlist secret
INGEST_URL = "https://backend.deml.app/api/v1/ingest"

response = redshift.execute_statement(
    ClusterIdentifier="prod-analytics",
    Database="analytics",
    Sql="SELECT feature_a, feature_b, label FROM ml.training_features LIMIT 1000",
)
statement_id = response["Id"]

# Poll until FINISHED, then fetch results and POST to DEML
records = [{"feature_a": 1.0, "feature_b": 0.5, "label": 1}]  # map from GetStatementResult
requests.post(
    INGEST_URL,
    headers={"Authorization": f"Bearer {API_KEY}"},
    json={"source": "redshift-data-api", "records": records},
    timeout=60,
).raise_for_status()

COPY from S3 After DEML Predictions

Write inference results back to the warehouse for BI dashboards:

COPY analytics.model_predictions
FROM 's3://your-bucket/deml-predictions/'
IAM_ROLE 'arn:aws:iam::123456789012:role/RedshiftCopyRole'
FORMAT AS JSON 'auto'
TIMEFORMAT 'auto';

Fetch predictions from DEML first:

curl -X POST https://backend.deml.app/api/v1/predict \
  -H "Authorization: Bearer YOUR_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"model_version": "v2", "inputs": [0.5, 0.2, 0.9]}'

Secrets Manager Pattern

Store the DEML API key alongside Redshift credentials:

import json
import boto3

secrets = boto3.client("secretsmanager")
payload = secrets.get_secret_value(SecretId="deml/production/api-key")
api_key = json.loads(payload["SecretString"])["DEML_API_KEY"]  # pragma: allowlist secret

Integration Health Check

curl https://backend.deml.app/api/v1/integrations/redshift \
  -H "Authorization: Bearer YOUR_API_KEY"

Expected response:

{
  "integration": "AWS Redshift",
  "status": "ready",
  "enabled": true,
  "version": "2.0+",
  "message": "AWS Redshift warehouse integration is active."
}

Related Guides

• Apache Spark — lakehouse batch and streaming sinks
• Databricks — notebook and job scheduling on AWS
• PyTorch — train on features exported from Redshift

TensorFlow Integration

Stream training data from DEML directly into a tf.data.Dataset for batched, high-throughput TensorFlow training loops.

Prerequisites

• Python 3.11+
• TensorFlow 2.15+
• A DEML API key (Settings → API Keys in the dashboard)

Quick Start

Install the SDK (planned package)

pip install deml-tensorflow

Until the package ships, use the REST ingest endpoint with a custom generator (see below).

Stream via tf.data.Dataset

import json
import tensorflow as tf
import requests

API_KEY = "YOUR_API_KEY"  # pragma: allowlist secret
INGEST_URL = "https://backend.deml.app/api/v1/ingest"
PREDICT_URL = "https://backend.deml.app/api/v1/predict"


def fetch_batch(batch_size: int = 32) -> list[dict]:
    response = requests.post(
        INGEST_URL,
        headers={"Authorization": f"Bearer {API_KEY}"},
        json={"batch_size": batch_size, "format": "tensorflow"},
        timeout=30,
    )
    response.raise_for_status()
    return response.json()["records"]


def record_generator():
    while True:
        for record in fetch_batch():
            yield record["features"], record["label"]


def build_dataset(batch_size: int = 32) -> tf.data.Dataset:
    dataset = tf.data.Dataset.from_generator(
        record_generator,
        output_signature=(
            tf.TensorSpec(shape=(None,), dtype=tf.float32),
            tf.TensorSpec(shape=(), dtype=tf.int32),
        ),
    )
    return dataset.batch(batch_size).prefetch(tf.data.AUTOTUNE)


model = tf.keras.Sequential([
    tf.keras.layers.Dense(64, activation="relu"),
    tf.keras.layers.Dense(1, activation="sigmoid"),
])

model.compile(optimizer="adam", loss="binary_crossentropy", metrics=["accuracy"])
model.fit(build_dataset(), steps_per_epoch=100, epochs=10)

Real-time Inference

Call /api/v1/predict from TensorFlow Serving sidecars or directly in training callbacks:

import requests

payload = {"model_version": "v2", "inputs": [0.5, 0.2, 0.9]}
result = requests.post(
    PREDICT_URL,
    headers={"Authorization": f"Bearer {API_KEY}"},
    json=payload,
    timeout=5,
)
prediction = result.json()["outputs"]

Planned SDK API

When deml-tensorflow ships, the interface will simplify to:

from deml.tensorflow import PlatformDataset

dataset = PlatformDataset(
    api_key="YOUR_API_KEY",  # pragma: allowlist secret
    batch_size=32,
    prefetch=True,
)
model.fit(dataset, epochs=10)

Integration Health Check

curl https://backend.deml.app/api/v1/integrations/tensorflow \
  -H "Authorization: Bearer YOUR_API_KEY"