Programarea si Orchestrarea Job-urilor Talend: Ghid Complet TAC

#!/bin/bash
# deploy_job.sh - Deploy job Talend in TAC
 
# Configurare
TAC_URL="http://tac-server:8080/org.talend.administrator"
TAC_USER="admin"
TAC_PASSWORD="password"  # Foloseste metoda securizata in productie
PROJECT_NAME="MY_PROJECT"
JOB_NAME="Daily_ETL_Job"
JOB_VERSION="1.0"
 
# Build arhiva job (din Talend Studio sau CI/CD)
# Aceasta creeaza un fisier .zip cu job-ul compilat
 
# Deploy folosind TAC REST API
curl -X POST "${TAC_URL}/rest/deployment/deployJob" \
  -H "Content-Type: multipart/form-data" \
  -u "${TAC_USER}:${TAC_PASSWORD}" \
  -F "file=@${JOB_NAME}_${JOB_VERSION}.zip" \
  -F "project=${PROJECT_NAME}" \
  -F "jobName=${JOB_NAME}" \
  -F "jobVersion=${JOB_VERSION}" \
  -F "context=Production" \
  -F "overwrite=true"
 
echo "Job deployed: ${JOB_NAME} v${JOB_VERSION}"

<!-- Configurare Talend JobServer -->
<!-- Locatie: {TALEND_HOME}/conf/TalendJobServer.properties -->
 
# Identificare server
talend.job.server.name=JobServer-ETL-01
talend.job.server.description=Primary ETL Job Server
 
# Setari conexiune
talend.job.server.port=8888
talend.job.server.host=0.0.0.0
 
# Setari executie
talend.job.server.concurrent.jobs=10
talend.job.server.timeout=3600000
 
# Setari memorie
talend.job.server.java.options=-Xms1g -Xmx4g -XX:+UseG1GC
 
# Setari log
talend.job.server.log.level=INFO
talend.job.server.log.path=/var/log/talend/jobserver
 
# Securitate
talend.job.server.ssl.enabled=true
talend.job.server.ssl.keystore=/path/to/keystore.jks
talend.job.server.ssl.keystore.password=${KEYSTORE_PASSWORD}

/*
 * Sintaxa expresiei Cron TAC:
 * Secunde Minute Ore ZiLuna Luna ZiSaptamana An
 *
 * Exemple:
 * 0 0 6 * * ?       - In fiecare zi la 6:00 AM
 * 0 0 */4 * * ?     - La fiecare 4 ore
 * 0 30 8 ? * MON-FRI - Zilele lucratoare la 8:30 AM
 * 0 0 0 1 * ?       - Prima zi din fiecare luna la miezul noptii
 * 0 0 22 ? * SUN    - In fiecare duminica la 10:00 PM
 */
 
// Pattern-uri comune de programare:
 
// ETL zilnic la 2 AM (ore de varf scazut)
String dailyEtlCron = "0 0 2 * * ?";
 
// Incarcari incrementale la ora (exceptand miezul noptii)
String hourlyCron = "0 0 1-23 * * ?";
 
// La fiecare 15 minute in orele de program
String frequentCron = "0 0/15 8-18 ? * MON-FRI";
 
// Procesare de sfarsit de zi (zile lucratoare la 11 PM)
String eodCron = "0 0 23 ? * MON-FRI";
 
// Raport saptamanal luni dimineata
String weeklyCron = "0 0 7 ? * MON";
 
// Procesare de inchidere lunara (ultima zi din luna la 11 PM)
String monthlyCron = "0 0 23 L * ?";
 
// Procesare trimestriala (prima zi din Ian, Apr, Iul, Oct)
String quarterlyCron = "0 0 1 1 1,4,7,10 ?";

/*
 * Configurare File Trigger TAC:
 *
 * Trigger Type: File
 * Directory: /data/incoming/sales
 * File Pattern: sales_*.csv
 * Polling Interval: 60 secunde
 * File Age (secunde): 30 (asigura ca fisierul este scris complet)
 * Move processed files: /data/archive/sales
 */
 
// Parametrii trigger-ului de fisier transmisi job-ului:
// - ${TRIGGER_FILE_PATH}   : Calea completa a fisierului trigger
// - ${TRIGGER_FILE_NAME}   : Numele fisierului trigger
// - ${TRIGGER_FILE_DIR}    : Directorul ce contine fisierul
// - ${TRIGGER_TIMESTAMP}   : Timestamp-ul trigger-ului
 
// In job, acceseaza parametrii trigger-ului:
String inputFile = context.TRIGGER_FILE_PATH;
String fileName = context.TRIGGER_FILE_NAME;
 
System.out.println("Processing file: " + inputFile);
 
// Dupa procesare, fisierul este mutat automat in arhiva
// pe baza configurarii trigger-ului

/*
 * TAC Event Triggers:
 *
 * 1. Web Service Trigger
 *    - Expune endpoint REST
 *    - Sistemele externe apeleaza pentru a declansa job-ul
 *
 * 2. JMS/Kafka Trigger
 *    - Asculta mesaje
 *    - Declanseaza job-ul la primirea mesajului
 *
 * 3. Database Trigger
 *    - Interogeaza tabela pentru inregistrari noi
 *    - Declanseaza job-ul cand gaseste inregistrari
 */
 
// Endpoint Web Service Trigger:
// POST http://tac-server:8080/org.talend.administrator/rest/execution/trigger
// Body: {"jobName": "ProcessOrder", "context": "Production", "params": {...}}
 
// Query trigger poll baza de date:
String pollQuery = "SELECT id, payload FROM trigger_queue " +
    "WHERE status = 'PENDING' ORDER BY created_at LIMIT 10";
 
// Dupa executia job-ului, actualizeaza inregistrarile trigger:
String updateQuery = "UPDATE trigger_queue SET status = 'PROCESSED', " +
    "processed_at = GETDATE() WHERE id IN (?)";

/*
 * Plan Executie: Daily_Sales_Pipeline
 *
 * Scop: Procesare end-to-end zilnica a datelor de vanzari
 *
 * Pasi:
 * 1. Extract_Sales_Data (paralel)
 * 2. Extract_Customer_Data (paralel cu pasul 1)
 * 3. Transform_Sales (depinde de 1 si 2)
 * 4. Load_Data_Warehouse (depinde de 3)
 * 5. Generate_Reports (depinde de 4)
 * 6. Send_Notifications (depinde de 5)
 */
 
// Structura XML Plan Executie (pentru API/import)
/*
<?xml version="1.0" encoding="UTF-8"?>
<executionPlan name="Daily_Sales_Pipeline" description="Pipeline procesare zilnica vanzari">
 
    <step id="1" name="Extract_Sales">
        <job name="Extract_Sales_Data" version="1.0" context="Production"/>
        <parallelWith>2</parallelWith>
    </step>
 
    <step id="2" name="Extract_Customers">
        <job name="Extract_Customer_Data" version="1.0" context="Production"/>
        <parallelWith>1</parallelWith>
    </step>
 
    <step id="3" name="Transform">
        <job name="Transform_Sales" version="1.0" context="Production"/>
        <dependsOn>1,2</dependsOn>
        <onError>STOP_PLAN</onError>
    </step>
 
    <step id="4" name="Load_DW">
        <job name="Load_Data_Warehouse" version="1.0" context="Production"/>
        <dependsOn>3</dependsOn>
        <onError>STOP_PLAN</onError>
    </step>
 
    <step id="5" name="Reports">
        <job name="Generate_Reports" version="1.0" context="Production"/>
        <dependsOn>4</dependsOn>
        <onError>CONTINUE</onError>
    </step>
 
    <step id="6" name="Notify">
        <job name="Send_Notifications" version="1.0" context="Production"/>
        <dependsOn>5</dependsOn>
        <onError>CONTINUE</onError>
    </step>
 
    <trigger type="CRON" expression="0 0 2 * * ?"/>
 
</executionPlan>
*/

/*
 * Tipuri de dependente in TAC:
 *
 * 1. Secvential: Job B ruleaza dupa ce Job A se termina
 * 2. Paralel: Job-urile A si B ruleaza simultan
 * 3. Conditional: Job B ruleaza doar daca Job A reuseste/esueaza
 * 4. Bazat pe date: Job B ruleaza cand Job A produce date specifice
 */
 
// Folosind tRunJob pentru dependente incorporate:
/*
 * Master_Job:
 *   tRunJob_1 (Extract) → OnSubjobOk → tRunJob_2 (Transform) → OnSubjobOk → tRunJob_3 (Load)
 *                       → OnSubjobError → tRunJob_Error (Alert)
 */
 
// Configurare tRunJob:
/*
 * Job: Extract_Sales_Data
 * Context: Production
 * Pass context to child: true
 * Wait for job to complete: true
 *
 * Context parameters override:
 * - process_date: context.process_date
 * - batch_id: context.batch_id
 */
 
// Verificarea statusului job-ului copil:
int childExitCode = (Integer) globalMap.get("tRunJob_1_CHILD_RETURN_CODE");
String childJobId = (String) globalMap.get("tRunJob_1_CHILD_JOB_PID");
 
if (childExitCode != 0) {
    System.err.println("Child job failed with code: " + childExitCode);
    // Gestioneaza esecul
}

/*
 * Pattern-uri de executie paralela in TAC:
 *
 * 1. Paralelism la nivel de plan: Mai multi pasi ruleaza concurent
 * 2. Paralelism la nivel de job: Un singur job ruleaza instante multiple
 * 3. Paralelism la nivel de date: Job-ul proceseaza datele in thread-uri paralele
 */
 
// La nivel de plan: Pasii 1 si 2 marcati ca paraleli
// TAC UI: Bifeaza optiunea "Run in parallel with"
 
// La nivel de job: Instante multiple ale aceluiasi job
// Foloseste parametri diferiti pentru fiecare instanta:
/*
 * Instanta 1: region=US
 * Instanta 2: region=EU
 * Instanta 3: region=APAC
 */
 
// tParallelize in job pentru fluxuri de date paralele:
/*
 * tParallelize
 *   ├── Branch 1: Proceseaza tip fisier A
 *   ├── Branch 2: Proceseaza tip fisier B
 *   └── Branch 3: Proceseaza tip fisier C
 *
 * Toate ramurile se executa concurent
 */
 
// Punct de sincronizare dupa executia paralela:
/*
 * tParallelize branches ──► tSleep (punct sync) ──► tRunJob (pasul urmator)
 *
 * tSleep asteapta toate ramurile paralele sa se termine
 */

// TAC REST API - Obtine statusul executiei
/*
 * GET /rest/execution/status/{taskId}
 *
 * Raspuns:
 * {
 *   "taskId": "12345",
 *   "jobName": "Daily_ETL",
 *   "status": "RUNNING",
 *   "startTime": "2024-01-15T02:00:00Z",
 *   "currentStep": "Transform_Sales",
 *   "progress": 65,
 *   "rowsProcessed": 1500000
 * }
 */
 
// Query-uri dashboard monitorizare personalizat:
 
// Job-uri in curs de rulare:
String runningJobsQuery = "SELECT j.name, e.start_time, e.status, s.name as server " +
    "FROM executions e " +
    "JOIN jobs j ON e.job_id = j.id " +
    "JOIN servers s ON e.server_id = s.id " +
    "WHERE e.status = 'RUNNING'";
 
// Istoric executii job-uri:
String historyQuery = "SELECT j.name, e.start_time, e.end_time, e.status, " +
    "DATEDIFF(SECOND, e.start_time, e.end_time) as duration_sec " +
    "FROM executions e " +
    "JOIN jobs j ON e.job_id = j.id " +
    "WHERE e.start_time >= DATEADD(DAY, -7, GETDATE()) " +
    "ORDER BY e.start_time DESC";
 
// Job-uri esuate in ultimele 24 ore:
String failedJobsQuery = "SELECT j.name, e.start_time, e.error_message " +
    "FROM executions e " +
    "JOIN jobs j ON e.job_id = j.id " +
    "WHERE e.status = 'ERROR' " +
    "AND e.start_time >= DATEADD(DAY, -1, GETDATE())";

// Configurare alerte TAC prin API
/*
 * POST /rest/alerts/create
 *
 * {
 *   "name": "Critical Job Failure Alert",
 *   "type": "JOB_FAILURE",
 *   "jobs": ["Daily_ETL", "Monthly_Close"],
 *   "severity": "CRITICAL",
 *   "notification": {
 *     "email": ["data-team@company.com", "oncall@company.com"],
 *     "slack": "#data-alerts",
 *     "pagerduty": "P12345"
 *   },
 *   "conditions": {
 *     "consecutiveFailures": 1,
 *     "timeWindow": "1h"
 *   }
 * }
 */
 
// Template alerta email:
/*
 * Subject: [ALERT] Talend Job Failed: ${jobName}
 *
 * Job Failure Alert
 * =================
 *
 * Job Name: ${jobName}
 * Execution ID: ${executionId}
 * Start Time: ${startTime}
 * End Time: ${endTime}
 * Duration: ${duration}
 * Status: FAILED
 *
 * Error Message:
 * ${errorMessage}
 *
 * Server: ${serverName}
 * Environment: ${environment}
 *
 * Link to logs: ${tacUrl}/execution/${executionId}/logs
 */
 
// Alerta webhook Slack:
public static void sendSlackAlert(String webhookUrl, String jobName,
        String errorMessage, String executionId) throws Exception {
 
    String payload = String.format(
        "{\"attachments\": [{" +
        "\"color\": \"danger\"," +
        "\"title\": \"Talend Job Failed: %s\"," +
        "\"text\": \"%s\"," +
        "\"fields\": [" +
        "  {\"title\": \"Execution ID\", \"value\": \"%s\", \"short\": true}," +
        "  {\"title\": \"Time\", \"value\": \"%s\", \"short\": true}" +
        "]" +
        "}]}",
        jobName, escapeJson(errorMessage), executionId,
        new java.text.SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new java.util.Date())
    );
 
    java.net.URL url = new java.net.URL(webhookUrl);
    java.net.HttpURLConnection conn = (java.net.HttpURLConnection) url.openConnection();
    conn.setRequestMethod("POST");
    conn.setRequestProperty("Content-Type", "application/json");
    conn.setDoOutput(true);
    conn.getOutputStream().write(payload.getBytes("UTF-8"));
    conn.getResponseCode();
}

// Monitoreaza SLA-urile job-urilor (Service Level Agreements)
 
/*
 * Configurare SLA:
 * - Job: Daily_ETL
 * - Finalizare asteptata: Pana la 6:00 AM
 * - Durata maxima: 4 ore
 * - Alerteaza daca: Nu este finalizat pana la 5:30 AM (30 min avertisment)
 */
 
// tJava - Job verificare SLA
import java.time.*;
 
String jobName = "Daily_ETL";
LocalTime slaDeadline = LocalTime.of(6, 0);  // 6:00 AM
LocalTime warningTime = LocalTime.of(5, 30); // 5:30 AM avertisment
Duration maxDuration = Duration.ofHours(4);
 
// Verifica daca job-ul s-a terminat azi
boolean jobCompleted = checkJobCompletedToday(jobName);
LocalTime now = LocalTime.now();
 
if (!jobCompleted) {
    if (now.isAfter(slaDeadline)) {
        // SLA incalcat!
        sendAlert("CRITICAL", "SLA BREACH: " + jobName + " not completed by deadline");
 
    } else if (now.isAfter(warningTime)) {
        // Avertisment
        sendAlert("WARNING", "SLA at risk: " + jobName + " not yet completed");
    }
}
 
// Verifica durata job-ului in curs
Duration currentDuration = getJobRunningDuration(jobName);
if (currentDuration != null && currentDuration.compareTo(maxDuration) > 0) {
    sendAlert("WARNING", jobName + " running longer than expected: " +
        currentDuration.toHours() + " hours");
}

// Genereaza job-uri dinamic pe baza datelor
 
// Job-ul master citeste configuratia si lanseaza job-uri copil
/*
 * tDBInput (tabela job_config)
 *     │
 *     ▼
 * tFlowToIterate
 *     │
 *     ▼
 * tRunJob (nume job dinamic din config)
 */
 
// Tabela de configurare job-uri:
/*
 * CREATE TABLE etl_job_config (
 *     config_id INT PRIMARY KEY,
 *     job_name VARCHAR(100),
 *     source_system VARCHAR(50),
 *     source_table VARCHAR(100),
 *     target_table VARCHAR(100),
 *     enabled BIT,
 *     priority INT,
 *     parameters TEXT
 * );
 */
 
// tRunJob cu nume job dinamic:
/*
 * Job: (String) globalMap.get("row1.job_name")
 * Context: Production
 *
 * Context parameters:
 * - source_system: (String) globalMap.get("row1.source_system")
 * - source_table: (String) globalMap.get("row1.source_table")
 * - target_table: (String) globalMap.get("row1.target_table")
 */

#!/bin/bash
# promote_job.sh - Promoveaza job din DEV in PROD
 
DEV_TAC_URL="http://tac-dev:8080"
PROD_TAC_URL="http://tac-prod:8080"
JOB_NAME="$1"
JOB_VERSION="$2"
 
# Export din DEV
echo "Exporting ${JOB_NAME} v${JOB_VERSION} from DEV..."
curl -X GET "${DEV_TAC_URL}/rest/deployment/exportJob" \
  -u "admin:password" \
  -d "jobName=${JOB_NAME}" \
  -d "version=${JOB_VERSION}" \
  -o "${JOB_NAME}_${JOB_VERSION}.zip"
 
# Import in PROD (cu workflow de aprobare in scenariul real)
echo "Importing to PROD..."
curl -X POST "${PROD_TAC_URL}/rest/deployment/deployJob" \
  -u "admin:password" \
  -F "file=@${JOB_NAME}_${JOB_VERSION}.zip" \
  -F "overwrite=false"
 
# Actualizeaza contextul la Production
echo "Updating context..."
curl -X PUT "${PROD_TAC_URL}/rest/tasks/updateContext" \
  -u "admin:password" \
  -H "Content-Type: application/json" \
  -d "{\"jobName\": \"${JOB_NAME}\", \"context\": \"Production\"}"
 
echo "Promotion complete"

// Implementeaza capabilitate de restart bazata pe checkpoint
 
// La inceputul job-ului, verifica rularea anterioara incompleta
String jobId = jobName + "_" + context.batch_date;
String checkpointKey = "checkpoint_" + jobId;
 
// Citeste ultimul checkpoint
String lastCheckpoint = readCheckpoint(checkpointKey);
 
if (lastCheckpoint != null) {
    System.out.println("Resuming from checkpoint: " + lastCheckpoint);
    globalMap.put("resume_from", lastCheckpoint);
    globalMap.put("is_restart", true);
} else {
    System.out.println("Starting fresh run");
    globalMap.put("is_restart", false);
}
 
// In procesarea principala, salveaza checkpoint-uri periodic
// tJavaRow - Salveaza checkpoint la fiecare N inregistrari
int checkpointInterval = 10000;
int currentRow = ((Integer) globalMap.get("row_counter", 0)) + 1;
globalMap.put("row_counter", currentRow);
 
if (currentRow % checkpointInterval == 0) {
    saveCheckpoint(checkpointKey, String.valueOf(input_row.id));
    System.out.println("Checkpoint saved at row " + currentRow);
}
 
// La sfarsitul job-ului, sterge checkpoint-ul la succes
// tJava (OnSubjobOk)
deleteCheckpoint(checkpointKey);
System.out.println("Job completed, checkpoint cleared");

scheduling_best_practices:
  design:
    - "Foloseste planuri de executie pentru pipeline-uri multi-job"
    - "Implementeaza dependente corecte intre job-uri"
    - "Proiecteaza pentru capabilitate de restart/recuperare"
    - "Foloseste nume semnificative pentru job-uri si planuri"
 
  programare:
    - "Programeaza job-urile intensive in ore de varf scazut"
    - "Etaleaza timpii de start pentru a evita contentia"
    - "Lasa timp buffer intre job-urile dependente"
    - "Ia in considerare diferentele de fus orar pentru operatii globale"
 
  monitorizare:
    - "Configureaza alerte pentru toate job-urile critice"
    - "Monitoreaza SLA-urile proactiv"
    - "Urmareste tendintele duratei job-urilor"
    - "Revizuieste prompt log-urile job-urilor esuate"
 
  mentenanta:
    - "Arhiveaza log-urile vechi de executie regulat"
    - "Revizuieste si curata job-urile neutilizate"
    - "Testeaza periodic procedurile de recuperare"
    - "Documenteaza deciziile de programare"
 
  securitate:
    - "Foloseste control al accesului bazat pe roluri"
    - "Cripteaza parametrii de context sensibili"
    - "Auditeaza toate executiile job-urilor"
    - "Separa mediile corect"