We are excited to announce the release of our Platform Streamsets SDK 7.0. This update focuses on improving developer productivity, reliability, and consistency across the SDK.
Python Code Generator
The Python code generator automatically converts existing pipeline definitions into executable Python SDK code, enabling easy migration and code generation workflows.
This feature supports multiple input sources including Pipeline objects, .zip archives exported from the UI, and .json files. Whether you're migrating existing pipelines to code or want to learn SDK patterns from existing pipelines, the code generator provides clean, readable Python code that follows SDK best practices.
Example: Generate code from a Pipeline object
from streamsets.sdk import ControlHub
from streamsets.sdk.codegen import PythonGenerator
# Connect to Control Hub
sch = ControlHub(credential_id='your_cred_id', token='your_token')
# Get an existing pipeline
pipeline = sch.pipelines.get(commit_id='your_pipeline_id')
# Generate Python code from the pipeline
generator = PythonGenerator(
source=pipeline,
destination="/tmp/my_pipeline.py"
)
generator.save()
Example: Generate code from an exported ZIP archive
from streamsets.sdk.codegen import PythonGenerator
# Generate code from a pipeline exported as a ZIP file
generator = PythonGenerator(
source="/path/to/exported/pipeline.zip",
destination="/tmp/regenerated_pipeline.py"
)
generator.save()
Example: Generate code with additional configuration values
from streamsets.sdk.codegen import PythonGenerator
# Use custom environment variable names for credentials
generator = PythonGenerator(
source="/path/to/pipeline.json",
destination="/tmp/output.py",
sch_credential_id="SCH_CRED_ID",
sch_token="SCH_TOKEN"
)
generator.save()
Sample output of generated Python code
This sample pipeline has two stages, Dev Raw Data Source and Trash
import os
from streamsets.sdk import ControlHub
sch = ControlHub(
os.getenv("SCH_CREDENTIAL_ID"),
os.getenv("SCH_TOKEN"),
aster_url="https://cloud.streamsets.com"
)
engine = sch.engines.get(id="your-engine-id")
pipeline_builder = sch.get_pipeline_builder(engine_type="COLLECTOR", engine_id=engine.id)
dev_raw_data_source_1 = pipeline_builder.add_stage("Dev Raw Data Source", type="origin")
dev_raw_data_source_1.stop_after_first_batch = True
trash_1 = pipeline_builder.add_stage("Trash", type="destination")
dev_raw_data_source_1.connect_outputs(stages=[trash_1])
pipeline = pipeline_builder.build("My Pipeline")
sch.publish_pipeline(pipeline)
HTTP Retry Management
The new HTTP Retry Management feature introduces a comprehensive retry mechanism with exponential backoff and jitter, automatically handling transient failures without requiring any code changes.
The retry module provides intelligent retry logic for common HTTP errors including rate limiting (429), server errors (5xx), and network timeouts. The system uses exponential backoff with jitter to prevent overwhelming servers and avoid thundering herd problems.
Automatic Retry with Configuration
from streamsets.sdk import ControlHub
from streamsets.sdk.retry import RetryConfig, RetrySettings
# Configure global retry behavior
RetryConfig.set(
default=RetrySettings(
max_attempts=5,
max_time=300, # 5 minutes maximum
init_delay=2.0,
exp_factor=2.5,
jitter=True
)
)
# All API calls now automatically retry on transient failures
pipeline = sch.pipelines.get(commit_id='my_pipeline')
sch.publish_pipeline(pipeline)
Temporary Retry Overrides for Specific Operations
from streamsets.sdk.retry import retry_on_http_error, RetrySettings
with retry_on_http_error(
settings=RetrySettings(max_attempts=10, init_delay=2.0),
only_status_codes={429} # Only retry rate limiting
):
job = sch.jobs.get(job_id='critical-job')
sch.start_job(job)
View Current Retry Configuration
from streamsets.sdk.retry import RetryConfig
RetryConfig.show()
======================================================================
RETRY CONFIGURATION
STATUS: RETRIES ENABLED
======================================================================
Default Settings:
Applied to status codes: [500, 502, 503, 504]
Max Attempts: 3
Max Time: None (max allowed: 3600s)
Initial Delay: 1.0s
Exp Factor: 2.0x
Jitter: Enabled
Status-Specific Settings:
HTTP 429:
Max Attempts: 10
Max Time: None (max allowed: 3600s)
Initial Delay: 1.0s
Exp Factor: 1.5x
Jitter: Enabled
======================================================================
Engine Type Standardization
The new EngineType enumeration centralizes all engine type references into a single, consistent interface. All engine types are now standardized as COLLECTOR, TRANSFORMER, and SNOWPARK across the entire SDK, while maintaining backward compatibility with legacy string values.
from streamsets.sdk import ControlHub
from streamsets.sdk.constants import COLLECTOR, TRANSFORMER, SNOWPARK
sch = ControlHub(credential_id='cred_id', token='token')
# Consistent engine type references
engine = sch.engines.get(engine_type=COLLECTOR)
pipeline_builder = sch.get_pipeline_builder(
engine_type=COLLECTOR, # Same constant everywhere
engine_id=engine.id
)
deployment = sch.get_deployment_builder().build(
engine_type=COLLECTOR, # Consistent!
deployment_name='my-deployment'
)
# Engine type attributes now return EngineType instances
print(engine.engine_type) # <EngineType.COLLECTOR: 'COLLECTOR'>
Additional Updates
Along with these major features, version 7.0 includes important updates:
-
Python 3.10+ Required: Support for Python 3.8 and 3.9 has been dropped. Please upgrade to Python 3.10 or higher before upgrading to SDK 7.0.
-
All methods and properties marked for deprecation in SDK 7.0 have been removed. See the migration guide in #for details on updating your code.
Get Started
Upgrade to StreamSets SDK 7.0 today to take advantage of these powerful new features:
pip install --upgrade streamsets
Learn More
For detailed documentation on each feature, visit our IBM StreamSets SDK 7.0 Release Documentation.
We're excited to see how these features improve your pipeline development workflows. As always, we welcome your feedback and contributions to make the StreamSets SDK even better!