aidp-streaming-kafka

name	aidp-streaming-kafka
description	Consume an OCI Streaming stream from an AIDP notebook via Spark structured streaming (Kafka-compat). Use when the user mentions OCI Streaming, Kafka on OCI, stream pool, structured streaming, or wants to read Kafka messages into Spark. Auth is SASL/PLAIN with an OCI auth token. Pattern matches the official Oracle AIDP sample.
allowed-tools	Read, Write, Edit, Bash

aidp-streaming-kafka — OCI Streaming via Spark structured streaming

Mirrors the official Oracle AIDP sample at oracle-samples/oracle-aidp-samples → data-engineering/ingestion/Streaming/StreamingFromOCIStreamingService.ipynb.

When to use

• User wants to consume an OCI Streaming stream (Kafka-compat) from an AIDP notebook.
• User mentions: "OCI Streaming", "Kafka on OCI", "stream pool", "structured streaming", "Kafka topic".

When NOT to use

• For batch reads of files in OCI Object Storage → standard spark.read.format("csv"|"parquet").load("oci://...") is fine without this skill.
• For other Kafka deployments (Confluent, MSK) — same Spark Kafka API works; just point bootstrap.servers at the right broker and skip the OCI-specific username format.

Prerequisites in the AIDP notebook

1. Spark Kafka connector on the cluster (spark-sql-kafka-0-10_<scala>:<spark> — AIDP's tpcds cluster has this).
2. Helpers on sys.path.
3. OCI Streaming stream pool OCID + region.
4. An OCI auth token (Profile → Auth tokens → Generate Token in the OCI console). 1-hour TTL — refresh before any job that runs longer than that.
5. A Volumes-mounted checkpoint location (/Volumes/<catalog>/<schema>/<volume>/_checkpoints/...). Do NOT use /Workspace/... — the streaming engine fails silently. The helper's validate_checkpoint_path() raises a clear ValueError if you try.

Auth: SASL/PLAIN with OCI auth token

import os
from oracle_ai_data_platform_connectors.streaming import (
    bootstrap_for_region, build_kafka_options_sasl_plain,
    validate_checkpoint_path,
)

# Bootstrap. Either generic-regional (default) or cell-prefixed (matches OCI
# Console's "messages-endpoint" shape — pick whichever your stream pool shows):
bootstrap = bootstrap_for_region(os.environ["OCI_REGION"])              # streaming.<region>...:9092
# bootstrap = bootstrap_for_region(os.environ["OCI_REGION"], cell=1)    # cell-1.streaming.<region>...:9092

opts = build_kafka_options_sasl_plain(
    bootstrap_servers=bootstrap,
    tenancy_name=os.environ["OCI_TENANCY_NAME"],     # display name, NOT OCID
    username=os.environ["OCI_USERNAME"],             # OCI user; for IAM-Domains
                                                     # use "oracleidentitycloudservice/<email>"
    stream_pool_ocid=os.environ["OCI_STREAM_POOL_OCID"],
    auth_token=os.environ["OCI_AUTH_TOKEN"],         # 1h TTL — refresh before long jobs
    topic=os.environ["KAFKA_TOPIC"],
    starting_offsets="latest",                       # or "earliest" for backfill
    # Optional tuning (matches the official sample):
    max_partition_fetch_bytes=1024 * 1024,
    max_offsets_per_trigger=5,                       # cap rows per micro-batch (demo-friendly)
)

raw = spark.readStream.format("kafka").options(**opts).load()

# Validate checkpoint path BEFORE starting (saves you from silent FUSE failures)
checkpoint = validate_checkpoint_path(os.environ["KAFKA_CHECKPOINT_VOLUME"])
sink_path  = os.environ["KAFKA_SINK_VOLUME"]   # e.g. /Volumes/default/default/streaming/kafkaStreamingSink

# Match the official sample: write to a Delta sink under /Volumes/.
query = (
    raw.writeStream
       .queryName("OCIStreamingSource")
       .format("delta")
       .option("checkpointLocation", checkpoint)
       .start(sink_path)
)
query.awaitTermination(timeout=120)
print("input rows in last batch:", (query.lastProgress or {}).get("numInputRows"))

For an inline test against an existing topic with print-style output:

out_df = raw.selectExpr("CAST(key AS STRING) AS k", "CAST(value AS STRING) AS v",
                        "topic", "partition", "offset")
q = (out_df.writeStream.format("memory").queryName("kafka_test")
            .option("checkpointLocation", checkpoint)
            .trigger(processingTime="5 seconds").start())
q.awaitTermination(timeout=60)
spark.sql("SELECT * FROM kafka_test").show()
q.stop()

Username format (the most common gotcha)

OCI Streaming's Kafka SASL username is <tenancy_name>/<user>/<stream_pool_ocid>. The middle segment depends on tenancy type:

Tenancy	username argument
Legacy IAM	<email>
IAM Domains (modern)	oracleidentitycloudservice/<email>

If you oci iam user list shows the user with oracleidentitycloudservice/... prefix, use the prefixed form.

Gotchas

• Checkpoint path — must be /Volumes/.... The validate_checkpoint_path() helper raises a ValueError if you pass /Workspace/... or oci://.... This is the #1 cause of "stream runs but no data appears" complaints in AIDP.
• Auth token TTL = 1 hour. For longer runs, plan to checkpoint, stop the stream, refresh the token, restart from checkpoint. RP-based Kafka SASL (com.oracle.bmc.auth.sasl.ResourcePrincipalsLoginModule) is blocked at the AIDP platform level (RP tokens not provided).
• Username format — tenancy name (display name), NOT tenancy OCID. IAM-Domains users need the oracleidentitycloudservice/ prefix.
• Streaming jobs run forever. The AIDP workflow timeout doesn't apply once a streaming query is started. Set Max Concurrent Runs = 1 on the wrapping job.
• Bootstrap host — the OCI Console's stream-pool detail page shows a "messages-endpoint" like https://cell-1.streaming.<region>.oci.oraclecloud.com. Either form (streaming.<region>... or cell-N.streaming.<region>...) works for the Kafka layer.

References

• Helpers: scripts/oracle_ai_data_platform_connectors/streaming/kafka.py
• Official Oracle AIDP sample: StreamingFromOCIStreamingService.ipynb
• OCI Streaming Kafka compat docs: https://docs.oracle.com/en-us/iaas/Content/Streaming/Tasks/kafkacompatibility_topic-Configuration.htm