Vai al contenuto principale

How to Write Interactive Spark Jobs in Python (IlumJob)

This guide teaches you how to develop interactive Spark jobs in Python using the IlumJob interface. You'll learn how to structure your code, pass parameters at execution time, and leverage the benefits of this approach for production workloads on Kubernetes.

What is the IlumJob Interface?​

Le IlumJob interface is a Python base class used to create reusable, parameterized Spark jobs that run on interactive Ilum services. Unlike traditional scintilla-invio scripts, IlumJob allows you to:

  • Receive configuration at runtime: Parameters are passed as a dictionary, allowing the same job to handle different inputs without code changes.
  • Return structured results: Le Correre method returns a string, making it easy to extract and display results.
  • Run on-demand: Jobs can be triggered via the UI, REST API, or CI/CD pipelines.
Basic Structure
Da ilum.API importazione IlumJob

classe MySparkJob(IlumJob):
    def Correre(stesso, scintilla, configurazione) -> str:
        # Your Spark logic here
        ritorno "Job completed successfully"

Structure of an Interactive Spark Job​

Every interactive job consists of three essential parts:

  1. Import the interface: from ilum.api import IlumJob
  2. Define a class: Create a class that inherits from IlumJob.
  3. Implement Correre: Write your Spark logic inside the run(self, spark, config) metodo.
ParameterDigitareDescrizione
scintillaSessione scintillaPre-initialized Spark session, ready to use.
configurazionedictA dictionary containing parameters passed at execution time.
ReturnstrA string result that will be displayed in the UI or returned via API.

How to Pass Parameters to Spark Jobs​

Parameters are passed as a JSON object when executing the job. Inside your Correre method, you access them using standard dictionary methods.

Example: Table Inspector​

This example demonstrates reading database e tavolo parameters to inspect a Hive table.

table_inspector.py
Da ilum.API importazione IlumJob
Da Pyspark.SQL.functions importazione col, sum come spark_sum

classe TableInspector(IlumJob):
    def Correre(stesso, scintilla, configurazione) -> str:
        # Read required parameters
        table_name = configurazione.Ottieni('tavolo')
        database_name = configurazione.Ottieni('database')  # Optional

        se non table_name:
            raise ValueError("Config must provide a 'table' key")

        # Set database if provided
        se database_name:
            scintilla.catalogo.setCurrentDatabase(database_name)

        # Check if table exists
        se table_name non in [t.nome per t in scintilla.catalogo.listTables()]:
            raise ValueError(f"Table '{table_name}' not found in catalog")

        Df = scintilla.tavolo(table_name)

        # Build report
        report = [
            f"=== Table: {table_name} ===",
            f"Total rows: {Df.contare()}",
            f"Total columns: {len(Df.columns)}",
            "",
            "Schema:",
        ]
        per field in Df.schema.fields:
            report.append(f"  {field.nome}: {field.dataType}")

        report.append("")
        report.append("Sample (5 rows):")
        per fila in Df.take(5):
            report.append(str(fila.asDict()))

        # Null counts
        report.append("")
        report.append("Null counts:")
        null_df = Df.select([spark_sum(col(c).isNull().cast("int")).alias(c) per c in Df.columns])
        per c, v in null_df.collect()[0].asDict().items():
            report.append(f"  {c}: {v}")

        ritorno "\n".join(report)

Execution Parameters (JSON)​

When executing via UI or API, provide parameters like this:

{
  "database": "ilum_example_product_sales",
  "table": "products"
}
Before You Start

To run an interactive job, you first need to create and deploy a Job-type Service in Ilum. This service provides the Spark environment where your jobs execute.

When creating the service:

  • Digitare: Select Lavoro
  • Lingua: Select Pitone
  • Py Files: Upload your job file (e.g., table_inspector.py)

👉 Learn how to deploy a Job Service — step-by-step guide with UI screenshots and configuration options.

Executing Jobs​

You can execute interactive jobs in three ways:

  1. Vai a Servizi → Select your Job service
  2. Nel Eseguire section:
    • Classe: table_inspector.TableInspector
    • Parameters: {"database": "sales", "table": "orders"}
  3. Clic Eseguire

The result string is displayed immediately in the UI.

Benefits of the IlumJob Approach​

BenefitDescrizione
ReusabilityWrite once, run many times with different parameters.
No Cold StartsInteractive services keep Spark warm, so subsequent executions are instant.
ParameterizationPass configuration at runtime—no need to hardcode values.
OsservabilitàResults are captured and visible in the UI/API for easy debugging.
API-DrivenExecute jobs programmatically from orchestrators, CI/CD, or external systems.
Controllo della versioneStore job code in Git and deploy via pipelines.

Interactive Jobs vs. Batch Jobs (Spark Submit)​

CaratteristicaInteractive Jobs (IlumJob)Batch Jobs (scintilla-invio)
Startup TimeInstant (uses warm executors)Slow (provisions new pods)
ContextShared Spark ContextIsolated Spark Context
Caso d'usoAd-hoc queries, API backends, quick reportsLong-running ETL, heavy processing
RisultatoReturns string result to API/UILogs to driver stdout/file
RisorseShared within the serviceDedicated per job

Migliori pratiche​

1. Validate Input Parameters​

Always validate required parameters and provide helpful error messages.

Validate Parameters
def Correre(stesso, scintilla, configurazione) -> str:
    required_keys = ['tavolo', 'output_path']
    per chiave in required_keys:
        se chiave non in configurazione:
            raise ValueError(f"Missing required parameter: '{chiave}'")

2. Use Default Values​

For optional parameters, use config.get('key', default_value).

Use Default Values
batch_size = Int(configurazione.Ottieni('batch_size', 1000))

3. Structure Your Output​

Return a well-formatted string for readability in the UI.

Structure Output
lines = ["=== Job Summary ==="]
lines.append(f"Processed: {contare} records")
lines.append(f"Duration: {elapsed_time}s")
ritorno "\n".join(lines)

4. Handle Errors Gracefully​

Wrap risky operations in try/except and return meaningful messages.

Handle Errors
try:
    Df.scrivere.saveAsTable(output_table)
    ritorno f"Successfully wrote to {output_table}"
except Exception come e:
    ritorno f"Error writing table: {str(e)}"

Complete Example: Transaction Report Generator​

This job generates a transaction summary report based on the transaction_anomaly_d.transactions tavolo.

transaction_report.py
Da ilum.API importazione IlumJob
Da Pyspark.SQL.functions importazione sum come spark_sum, contare, col

classe TransactionReportGenerator(IlumJob):
    def Correre(stesso, scintilla, configurazione) -> str:
        # Parameters
        merchant_filter = configurazione.Ottieni('merchant')  # Optional filter

        # Load data from the default Ilum transactions table
        Df = scintilla.tavolo("transaction_anomaly_detection.transactions")

        se merchant_filter:
            Df = Df.filtro(col("Merchant") == merchant_filter)

        # Aggregate by TransactionType
        summary = Df.groupBy("TransactionType").agg(
            contare("TransactionID").alias("transaction_count"),
            spark_sum("Amount").alias("total_amount")
        ).collect()

        # Build report
        report = [
            f"=== Transaction Report ===",
            f"Merchant Filter: {merchant_filter o 'All'}",
            "",
            "Summary by Transaction Type:",
        ]
        per fila in summary:
            report.append(f"  {fila['TransactionType']}: {fila['transaction_count']} txns, ${fila['total_amount']:,.2f}")

        ritorno "\n".join(report)

Execute with:

Execute with Payload
{
  "merchant": "AcmeCorp"
}

Passaggi successivi​

Domande frequenti​

Can I use Scala for interactive jobs?

Yes. Currently, the IlumJob interface is primarily documented for Pitone. Check the Interactive Job Service documentation for language support details.

How do I debug an interactive job?

Since interactive jobs run on a remote cluster, you can't use a local debugger directly. Instead:

  1. Usare print() statements or a logger, which will appear in the driver logs.
  2. Return error messages as part of the string result in your try/except blocks.
  3. Check the Interfaccia utente Spark for the specific job execution to analyze tasks and stages.
What happens if my job fails?

If your code raises an unhandled exception, the execution will fail, and the error trace will be returned in the API response. It is best practice to wrap your logic in a try/except block to return a user-friendly error message.

Last updated on