Programma

Panoramica

Ilum Schedule consente di impostare e gestire facilmente le pianificazioni per i singoli lavori attraverso un'interfaccia intuitiva e facile da usare. Semplifica il processo di configurazione delle tempistiche di esecuzione con facilità.

Dietro le quinte, questa funzione sfrutta Kubernetes CronJobs, che utilizza espressioni CRON per definire con precisione quando i lavori devono essere eseguiti. All'ora pianificata, il CronJob attiva una richiesta al backend di Ilum, che orchestra la creazione di un singolo lavoro.

Casi d'uso

Acquisizione dei dati

Recupera periodicamente i dati da API esterne o altre fonti e caricali in un database o in un file system per un'ulteriore elaborazione o analisi.

Orchestrazione della pipeline di dati (ETL)

Automatizza i processi ETL (Extract, Transform, Load) basati su Spark che estraggono dati grezzi da più origini, applicano trasformazioni complesse e caricano il Risultati in un data warehouse

Preparazione di report e aggregazione dei dati per l'analisi

Pianificare un processo Spark per aggregare set di dati di grandi dimensioni da varie fonti (ad esempio, registri, dati di vendita, interazioni con gli utenti) in tabelle di riepilogo che vengono utilizzate in dashboard e report.

Pulizia dei dati

Pianificare un processo Spark che esegua la pulizia dei dati, ad esempio la rimozione di duplicati, la correzione di voci non valide o il filtraggio dei dati, in un set di dati di grandi dimensioni archiviato in un sistema distribuito

Inizia

Processo Python pianificato

L'esempio seguente è un semplice processo Spark in Python che genera dati casuali e li scrive in un bucket S3. L'esecuzione di questo processo verrà pianificata in un intervallo di tempo specificato.

Scrivere un singolo processo

Da Pyspark.SQL importazione Sessione scintilla
Da Pyspark.SQL importazione Fotogramma dati
Da Pyspark.SQL importazione Fila
importazione aleatorio
Da dataora importazione dataora


def generate_sample_data(n):
    ritorno [Fila(Id=io, nome=f"Name_{io}", valore=aleatorio.Randrange(70, 100)) per io in gamma(n)]


se __nome__ == "__main__":
    scintilla = Sessione scintilla.muratore\
    .nomeapp("Il mio lavoro scintillante") \
    .getOrCreate()

    dati = generate_sample_data(100)
    Df = scintilla.createDataFrame(dati)

    current_datetime = dataora.ora()
    current_timestamp = current_datetime.strftime("%Y-%m-%d %H:%M:%S")

    output_path = f"s3a://ilum-files/generated_data/{current_timestamp}"
    Df.scrivere.modo("sovrascrivere").parquet(output_path)

    scintilla.fermarsi()

Salva il file con nome .py

Creare una pianificazione

Quando crei la pianificazione, specifica quanto segue:
- Scheda Generale:
  - Nome: Nome del programma
  - Grappolo: il cluster in cui verrà eseguito il processo
  - Lingua: Python in questo caso
  - Classe: senza il .Py estensione
- Tabella Temporizzazione:
  - Scegli un valore adatto alle tue esigenze. È possibile utilizzare un'espressione CRON personalizzata se le opzioni predefinite non soddisfano i requisiti.
- Scheda Risorse:
  - PyFiles: Carica lo script Python
Osservare i registri dei lavori visualizzati

Dopo aver creato la pianificazione, questa è già attiva e avvierà i processi in base alle impostazioni specificate.

I singoli lavori avviati dalla pianificazione si comportano allo stesso modo dei normali singoli lavori Ilum: Ad esempio, puoi andare alla pagina Lavori per visualizzare l'elenco dei processi avviati dalle pianificazioni.

Processo Scala pianificato

Questo esempio è lo stesso di quello di Python, ma scritto in Scala.

Creare un progetto sbt

Durante la creazione del progetto, è importante includere il spark-sql dipendenza nel build.sbt file.
```
libraryDependencies += "org.apache.spark" %% "spark-sql" % "3.5.3" % "fornito"
```
Tenere presente che la versione della dipendenza deve corrispondere alla versione di Spark in esecuzione nel cluster.

Scrivere e compilare un singolo lavoro

importazione org.apache.scintilla.SQL.types.{IntegerType, StringType, StructField, StructType}
importazione org.apache.scintilla.SQL.{Fila, Sessione scintilla}

importazione java.time.LocalDateTime
importazione java.time.format.DateTimeFormatter
importazione Scala.util.Random

oggetto Principale {
  private def generateSampleData(n: Int): Seq[Fila] = {
    (0 until n).mappa(io => Fila(io, s"Name_$io", Random.nextInt(30) + 70))
  }

  def principale(argomenti: Array[String]): Unit = {
    val scintilla = Sessione scintilla.builder()
      .nomeapp("Il mio lavoro scintillante")
      .getOrCreate()

    val dati = generateSampleData(100)
    val Df = scintilla.createDataFrame(
      scintilla.sparkContext.parallelizzare(dati),
      StructType(
        List(
          StructField("id", IntegerType, nullable = falso),
          StructField("nome", StringType, nullable = falso),
          StructField("value", IntegerType, nullable = falso)
        )
      )
    )

    val currentDateTime = LocalDateTime.ora()
    val currentTimestamp = currentDateTime.format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"))

    val outputPath = s"s3a://ilum-files/generated_data/$currentTimestamp"
    Df.scrivere.modo("sovrascrivere").parquet(outputPath)

    scintilla.fermarsi()
  }
}

Ora il progetto può essere compilato in un file JAR. Mentre normalmente si vorrebbe un 'fat JAR' con tutte le dipendenze incluse, Questo processo è abbastanza semplice da poter essere eseguito con le dipendenze fornite dal cluster.

Per impacchettare rapidamente il progetto, eseguire il comando Pacchetto SBT comando.

Creare una pianificazione

Quando crei la pianificazione, specifica quanto segue:
- Scheda Generale:
  - Nome: Nome del programma
  - Grappolo: il cluster in cui verrà eseguito il processo
  - Lingua: Scala in questo caso
  - Classe: nome della classe canonica. Nel nostro caso sbt dovrebbe mettere il file di classe nella radice del JAR, quindi Principale dovrebbe essere sufficiente.
- Tabella Temporizzazione:
  - Scegli un valore adatto alle tue esigenze. È possibile utilizzare un'espressione CRON personalizzata se le opzioni predefinite non soddisfano i requisiti.
- Scheda Risorse:
  - Vasi: Carica il file JAR
Osservare i registri dei lavori visualizzati

Dopo aver creato la pianificazione, questa è già attiva e avvierà i processi in base alle impostazioni specificate.

I singoli lavori avviati dalla pianificazione si comportano allo stesso modo dei normali singoli lavori Ilum: Ad esempio, puoi andare alla pagina Lavori per visualizzare l'elenco dei processi avviati dalla pianificazione.

Esempio di pianificazione del caricamento

Ilum fornisce un esempio di pianificazione per aiutare i nuovi utenti a iniziare rapidamente. Esempio di caricamento della pianificazione è abilitato per impostazione predefinita. Tuttavia, è possibile disabilitarlo utilizzando --set ilum-core.examples.schedule=false.

Suggerimenti

Sfrutta tutta la potenza delle espressioni CRON

Sebbene Ilum sia dotato di opzioni predefinite per la pianificazione, è anche possibile utilizzare espressioni CRON personalizzate per impostare pianificazioni più complesse.

L'espressione Cron è composta da 5 campi:
    

* - ogni unità di tempo
? - qualsiasi unità di tempo (? nel campo del giorno della settimana comporta l'ignoranza del giorno della settimana)
- - gamma (1-5)
, - valori: (1,5)
/ - incrementi (5/15 nel campo dei minuti = 5, 20, 35, 50)

Esempi:

Ogni ora:
0 * * * *

Ogni domenica alle 3 del mattino:
0 3 * * 0

Ogni 15 minuti ogni giorno:
0/15 0 * * ?

Ogni cinque minuti a partire dalle 13:00, fino alle 13:55 e poi a partire dalle 18:00 e terminando alle 18:55, ogni giorno:
0/5 13,18 * * ?

Se vuoi saperne di più, dai un'occhiata al Documentazione Cron Job di Kubernetes.

Testa il tuo lavoro prima di pianificare

I lavori pianificati vengono avviati periodicamente, quindi è essenziale assicurarsi che il lavoro funzioni correttamente. Poiché i processi pianificati si comportano allo stesso modo dei normali processi singoli, È possibile testare l'esecuzione avviando un singolo processo prima di creare una pianificazione.

Pianifica avvia singoli lavori Ilum

I processi pianificati vengono avviati come processi singoli Ilum. Ciò significa che tutte le configurazioni, i registri, gli output e le altre funzionalità disponibili per i singoli processi sono disponibili anche per i processi pianificati.

Disattiva le pianificazioni quando non sono necessarie

Se non è necessario che il processo venga avviato periodicamente, è necessario disattivare le pianificazioni per evitare sprechi di risorse. Per fare ciò, puoi fare clic sul pulsante Pausa pulsante nel pulsante Orari scheda.

Modifica le tue pianificazioni invece di crearne una nuova

Se si desidera modificare le configurazioni di una pianificazione esistente, fare clic su Redigere su di esso.

Panoramica​

Casi d'uso​

Acquisizione dei dati​

Orchestrazione della pipeline di dati (ETL)​

Preparazione di report e aggregazione dei dati per l'analisi​

Pulizia dei dati​

Inizia​

Processo Python pianificato​

Processo Scala pianificato​

Esempio di pianificazione del caricamento​

Suggerimenti​

Sfrutta tutta la potenza delle espressioni CRON​

Testa il tuo lavoro prima di pianificare​

Pianifica avvia singoli lavori Ilum​

Disattiva le pianificazioni quando non sono necessarie​

Modifica le tue pianificazioni invece di crearne una nuova​