StreamSets Python SDK

Warnings

• breaking StreamSets SDK for Python versions 4.0.0 and higher are designed for the StreamSets DataOps Platform, while versions below 4.0.0 support legacy StreamSets products. Ensure your SDK version matches the StreamSets product version you intend to interact with for compatibility.
  Fix: Verify and align your installed SDK version with your StreamSets product (Platform or Legacy). Upgrade SDK to >=4.0.0 for Platform or use <4.0.0 for Legacy products.
• breaking Starting with SDK v3.11.0, methods related to Topologies were moved from the `streamsets.sdk.ControlHub` class to `streamsets.sdk.sch_models.Topology`. Existing code using `control_hub.get_topology_by_name()` or similar directly on the `ControlHub` object will fail.
  Fix: Update imports and method calls to use `from streamsets.sdk.sch_models import Topology` and interact with `Topology` objects directly for topology management.
• breaking The SDK for Python v3.10.0 refactored SDC pipeline snapshots. Upgrading to 3.10.0 or later without modifying existing code to use the new snapshot syntax will result in execution failures.
  Fix: Refer to the updated documentation for `streamsets.sdk.sdc_models.Snapshot` to adjust your code to the new syntax for pipeline snapshot interactions.
• gotcha For StreamSets SDK versions that interact with Control Hub (typically 4.0.0+), authentication requires API credentials (`CRED_ID` and `CRED_TOKEN`), which must be generated in Control Hub. This replaces the username and password authentication methods used in older SDK v3.x versions.
  Fix: Generate API credentials (Credential ID and Token) in your StreamSets Control Hub instance and use them to initialize the `ControlHub` object, preferably via environment variables.
• gotcha When using the SDK, creating an object (e.g., a `PipelineBuilder` or a pipeline object) in your Python script does not automatically mean that object exists within StreamSets Control Hub or Data Collector. You must explicitly 'add' or 'publish' the object (e.g., `control_hub.publish_pipeline(pipeline)`) for it to be reflected in the StreamSets environment.

Install

• pip install streamsets Install latest version

Imports

• ControlHub

  from streamsets.sdk import ControlHub

  Main class for interacting with StreamSets Control Hub.
• DataCollector

  from streamsets.sdk import DataCollector

  Main class for interacting with StreamSets Data Collector instances.
• Topology

  from streamsets.sdk.sch_models import Topology

  As of SDK v3.11.0, Topology-related methods were moved from `ControlHub` to `streamsets.sdk.sch_models.Topology` for proper scope.

Quickstart

This quickstart demonstrates how to connect to StreamSets Control Hub using API credentials and list available Data Collectors. It also includes an example of how to initialize a `PipelineBuilder` to create an in-memory pipeline definition. Remember that creating an object in code does not automatically deploy it to StreamSets; explicit `publish` or `add` calls are required.

import os
from streamsets.sdk import ControlHub

# Ensure CRED_ID and CRED_TOKEN are set as environment variables
# Example: export CRED_ID='your_credential_id'
# Example: export CRED_TOKEN='your_credential_token'

cred_id = os.environ.get('CRED_ID', 'YOUR_CRED_ID')
cred_token = os.environ.get('CRED_TOKEN', 'YOUR_CRED_TOKEN')
sch_url = os.environ.get('SCH_URL', 'https://cloud.streamsets.com') # Or your on-prem Control Hub URL

# Connect to Control Hub
try:
    control_hub = ControlHub(sch_url=sch_url, credential_id=cred_id, credential_token=cred_token)
    print(f"Successfully connected to Control Hub at {sch_url}")

    # Example: List available Data Collectors
    data_collectors = control_hub.get_data_collectors()
    if data_collectors:
        print("Available Data Collectors:")
        for dc in data_collectors:
            print(f" - {dc.name} (ID: {dc.id})")
    else:
        print("No Data Collectors found.")

    # Example: Create a simple pipeline (requires a Data Collector if deployed)
    # This example demonstrates creating a pipeline in memory, but not deploying it.
    # For deployment, you'd typically use PipelineBuilder and then control_hub.publish_pipeline()
    # or control_hub.create_job() if managing via Control Hub.
    print("\nExample: Creating a simple in-memory pipeline builder (not yet deployed to Control Hub).")
    pipeline_builder = control_hub.get_pipeline_builder()
    dev_data_generator = pipeline_builder.add_stage('Dev Data Generator')
    trash = pipeline_builder.add_stage('Trash')
    dev_data_generator >> trash # Connect stages
    
    my_pipeline = pipeline_builder.build('My First SDK Pipeline')
    print(f"Created pipeline builder object: {my_pipeline.name}")
    # To deploy this pipeline, you would use control_hub.publish_pipeline(my_pipeline)

except Exception as e:
    print(f"Error connecting to Control Hub or performing operations: {e}")