Optimizarea Job-urilor Talend: Ghid Complet de Performance Tuning

#!/bin/bash
# talend_jvm_config.sh - Setari JVM optime pentru Talend
 
# Configurare standard de productie
export JAVA_OPTS="-Xms2g -Xmx8g \
    -XX:+UseG1GC \
    -XX:MaxGCPauseMillis=200 \
    -XX:+ParallelRefProcEnabled \
    -XX:+UseStringDeduplication \
    -XX:InitiatingHeapOccupancyPercent=45 \
    -XX:G1HeapRegionSize=16m \
    -XX:+HeapDumpOnOutOfMemoryError \
    -XX:HeapDumpPath=/var/log/talend/heapdump.hprof \
    -Dfile.encoding=UTF-8"
 
# Configurare pentru job-uri cu memorie mare
export JAVA_OPTS_LARGE="-Xms8g -Xmx32g \
    -XX:+UseG1GC \
    -XX:MaxGCPauseMillis=500 \
    -XX:ConcGCThreads=4 \
    -XX:ParallelGCThreads=8 \
    -XX:G1HeapRegionSize=32m \
    -XX:+UseStringDeduplication"
 
# Configurare cu latenta scazuta
export JAVA_OPTS_LOWLATENCY="-Xms4g -Xmx4g \
    -XX:+UseZGC \
    -XX:ZCollectionInterval=30 \
    -XX:+ZUncommit"

// tJavaRow - Procesare eficienta din perspectiva memoriei
// Evita stocarea datelor in colectii cand nu este necesar
 
// GRESIT: Acumulare in memorie
java.util.List<String> allRecords = new java.util.ArrayList<>();
allRecords.add(input_row.data); // Creste indefinit!
 
// CORECT: Proceseaza si elibereaza
output_row.processedData = transformData(input_row.data);
// Fara acumulare, memoria eliberata dupa fiecare rand
 
// Pattern de agregare eficient din perspectiva memoriei
// Foloseste tAggregateRow in loc de colectii manuale
// tAggregateRow foloseste spillover pe disc pentru seturi mari de date

<!-- Variabile de context job pentru optimizarea buffer-ului -->
<context>
    <!-- Dimensiuni buffer randuri -->
    <contextParameter name="BUFFER_SIZE" value="10000"/>
    <contextParameter name="COMMIT_INTERVAL" value="5000"/>
 
    <!-- Configurare cache lookup -->
    <contextParameter name="LOOKUP_CACHE_SIZE" value="100000"/>
    <contextParameter name="LOOKUP_CACHE_TYPE" value="LRU"/>
 
    <!-- Configurare fisiere temporare -->
    <contextParameter name="TEMP_DIR" value="/data/talend/temp"/>
    <contextParameter name="USE_DISK_TEMP" value="true"/>
</context>

// Structura job principal cu ramuri paralele
 
/*
 * Design Job:
 *
 * [tParallelize] ──┬──► Branch 1: Proceseaza Date Clienti ──►┐
 *                  ├──► Branch 2: Proceseaza Date Comenzi  ──►├──► [tSleep/Join]
 *                  └──► Branch 3: Proceseaza Date Produse  ──►┘
 *
 * Fiecare ramura ruleaza intr-un thread separat
 */
 
// Configurare context pentru paralelizare
// tParallelize Advanced Settings:
// - Number of parallel processes: 4
// - Wait for all processes to complete: true
 
// In tJava (inainte de tParallelize):
System.setProperty("talend.parallel.threads", context.PARALLEL_THREADS);
globalMap.put("startTime", System.currentTimeMillis());

// tJavaFlex - Partitioneaza datele pentru procesare paralela
 
// Start Code
int partitionCount = Integer.parseInt(context.PARTITION_COUNT);
int currentPartition = 0;
java.util.List<java.util.List<Object>> partitions = new java.util.ArrayList<>();
for (int i = 0; i < partitionCount; i++) {
    partitions.add(new java.util.ArrayList<>());
}
 
// Main Code
int partitionIndex = Math.abs(input_row.id.hashCode()) % partitionCount;
partitions.get(partitionIndex).add(input_row);
 
// End Code
for (int i = 0; i < partitions.size(); i++) {
    globalMap.put("partition_" + i, partitions.get(i));
}
System.out.println("Partitioned " + globalMap.get("tJavaFlex_NB_LINE") + " rows into " + partitionCount + " partitions");
 
// Alternativa: Foloseste componenta tPartitioner (Talend DI)
// - Hash partitioning pe coloane cheie
// - Range partitioning pentru date sortate
// - Round-robin pentru distributie uniforma

// tJavaRow cu operatii thread-safe
 
// Folosire ThreadLocal pentru connection pooling
private static ThreadLocal<java.sql.Connection> connectionHolder =
    new ThreadLocal<java.sql.Connection>() {
        @Override
        protected java.sql.Connection initialValue() {
            try {
                return java.sql.DriverManager.getConnection(
                    context.db_url,
                    context.db_user,
                    context.db_password
                );
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        }
    };
 
// Contor thread-safe
private static java.util.concurrent.atomic.AtomicLong processedCount =
    new java.util.concurrent.atomic.AtomicLong(0);
 
// In codul principal:
java.sql.Connection conn = connectionHolder.get();
long count = processedCount.incrementAndGet();
 
if (count % 10000 == 0) {
    System.out.println("Processed: " + count + " records");
}

// tDBOutput - Configurare bulk insert
 
/*
 * Component Settings:
 *
 * Action on data: Insert
 * Commit every: 10000 (ajusteaza in functie de dimensiunea transaction log)
 *
 * Advanced settings:
 * - Use batch mode: true
 * - Batch size: 5000
 * - Use field options: false (pentru performanta bulk)
 * - Support null in SQL statement: false (daca nu sunt asteptate null-uri)
 */
 
// Setari bulk specifice Oracle (tOracleOutput)
/*
 * Use Array Insert: true
 * Array Insert Size: 5000
 * Disable indexes during insert: true (pentru incarcari mari)
 * Use direct path insert: true
 * Enable parallel DML: true
 */
 
// Setari bulk specifice SQL Server (tMSSqlOutput)
/*
 * Use bulk insert: true
 * Bulk file path: /data/bulk/
 * Lock table: true
 * Check constraint: false
 * Fire triggers: false
 */

// tJava - Initializare connection pool
 
// Folosirea HikariCP pentru connection pooling
com.zaxxer.hikari.HikariConfig config = new com.zaxxer.hikari.HikariConfig();
config.setJdbcUrl(context.db_url);
config.setUsername(context.db_user);
config.setPassword(context.db_password);
config.setMaximumPoolSize(20);
config.setMinimumIdle(5);
config.setConnectionTimeout(30000);
config.setIdleTimeout(600000);
config.setMaxLifetime(1800000);
config.addDataSourceProperty("cachePrepStmts", "true");
config.addDataSourceProperty("prepStmtCacheSize", "250");
config.addDataSourceProperty("prepStmtCacheSqlLimit", "2048");
 
com.zaxxer.hikari.HikariDataSource dataSource =
    new com.zaxxer.hikari.HikariDataSource(config);
globalMap.put("connectionPool", dataSource);
 
// In componentele ulterioare, obtine conexiunea din pool:
com.zaxxer.hikari.HikariDataSource pool =
    (com.zaxxer.hikari.HikariDataSource) globalMap.get("connectionPool");
java.sql.Connection conn = pool.getConnection();
try {
    // Foloseste conexiunea
} finally {
    conn.close(); // Se intoarce in pool
}
 
// Cleanup la sfarsitul job-ului (tJava - End):
com.zaxxer.hikari.HikariDataSource pool =
    (com.zaxxer.hikari.HikariDataSource) globalMap.get("connectionPool");
if (pool != null) {
    pool.close();
}

-- Exemple de optimizare tDBInput
 
-- GRESIT: Selecteaza tot apoi filtreaza in Talend
SELECT * FROM large_table;
-- Apoi foloseste tFilterRow pentru filtrare
 
-- CORECT: Filtreaza la nivel de baza de date
SELECT id, name, amount, created_date
FROM large_table
WHERE status = 'ACTIVE'
  AND created_date >= DATEADD(day, -30, GETDATE())
  AND amount > 0;
 
-- Foloseste coloane indexate in clauza WHERE
-- Adauga covering index pentru query-urile folosite frecvent:
CREATE INDEX idx_large_table_lookup
ON large_table (status, created_date)
INCLUDE (id, name, amount);
 
-- Pentru join-uri pe tabele mari, foloseste tDBJoin cu query hint
SELECT /*+ USE_HASH(a, b) PARALLEL(4) */
    a.id, a.name, b.total_amount
FROM customers a
INNER JOIN orders b ON a.customer_id = b.customer_id
WHERE b.order_date >= :startDate;
 
-- Paginare pentru seturi de date foarte mari
SELECT * FROM (
    SELECT ROW_NUMBER() OVER (ORDER BY id) as rn, t.*
    FROM large_table t
    WHERE status = 'ACTIVE'
) sub
WHERE rn BETWEEN :startRow AND :endRow;

// Strategii de optimizare lookup tMap
 
/*
 * Optiuni Lookup Model:
 *
 * 1. Load Once (implicit)
 *    - Incarca intreaga tabela lookup in memorie
 *    - Cel mai bun pentru: Tabele lookup mici (<100K randuri)
 *    - Memorie: Mare
 *
 * 2. Reload at Each Row
 *    - Interogeaza baza de date pentru fiecare rand principal
 *    - Cel mai bun pentru: Lookup-uri foarte mari cu flux principal selectiv
 *    - Memorie: Mica, dar lent
 *
 * 3. Reload at Each Row (cache)
 *    - Stocheaza in cache rezultatele lookup
 *    - Cel mai bun pentru: Lookup-uri mari cu repetitie de chei
 *    - Memorie: Medie
 */
 
// Pentru lookup-uri mari, foloseste pattern-ul tCacheIn/tCacheOut:
// 1. Incarca datele lookup in fisier cache
// 2. Foloseste tCacheLookup in tMap
 
// Configurare tCacheIn:
// Cache type: hash
// Cache directory: /data/talend/cache
// Key columns: lookup_key

// Bune practici pentru expresii tMap
 
// GRESIT: Expresii complexe repetate
// In coloane multiple de output:
StringHandling.UPCASE(StringHandling.TRIM(row1.name))
 
// CORECT: Foloseste variabile pentru expresii complexe
// In sectiunea Variables:
var_cleanName = StringHandling.UPCASE(StringHandling.TRIM(row1.name))
// Apoi foloseste var_cleanName in output-uri multiple
 
// GRESIT: Parsarea datei de mai multe ori
// Output 1: TalendDate.parseDate("yyyy-MM-dd", row1.dateStr)
// Output 2: TalendDate.getPartOfDate("MONTH", TalendDate.parseDate("yyyy-MM-dd", row1.dateStr))
 
// CORECT: Parseaza o singura data in variabila
// Variable: var_date = TalendDate.parseDate("yyyy-MM-dd", row1.dateStr)
// Output 1: var_date
// Output 2: TalendDate.getPartOfDate("MONTH", var_date)
 
// Evita verificarile null in calea critica
// Pre-filtreaza null-urile cu tFilterRow inainte de tMap
 
// Foloseste comparatii primitive cand este posibil
row1.status_code == 1  // Mai rapid decat
row1.status.equals("ACTIVE")  // Comparatie string

// Strategii de join tMap
 
/*
 * Optimizarea Inner Join:
 * - Pune tabela mai mica ca lookup
 * - Foloseste "Die on error" pentru lookup-uri obligatorii (esueaza rapid)
 *
 * Left Outer Join:
 * - Foloseste "Match: All matches" doar cand este necesar
 * - "First match" este mai rapid pentru relatii 1:1
 *
 * Lookup-uri Multiple:
 * - Ordoneaza lookup-urile dupa selectivitate (cel mai selectiv primul)
 * - Foloseste link-uri reject pentru a gestiona eficient lookup-urile lipsa
 */
 
// Estimarea memoriei pentru hash join:
// Memorie = (randuri x dimensiune_medie_rand) + overhead
// Exemplu: 1M randuri x 100 bytes = ~100MB + 20% overhead
 
// Pentru join-uri foarte mari, considera:
// 1. Pre-sorteaza ambele seturi de date
// 2. Foloseste tSortMergeJoin in loc de tMap
// 3. Partitioneaza datele si join-eaza partitiile in paralel

// tFileInputDelimited - Configurare fisiere mari
 
/*
 * Advanced Settings:
 * - Die on error: false (continua la inregistrari invalide)
 * - Split record: false (decat daca campurile se intind pe mai multe linii)
 * - Uncompress files: true (pentru input .gz)
 *
 * Pentru fisiere foarte mari:
 * - Foloseste tFileInputDelimited cu limit/start rows
 * - Proceseaza in chunk-uri cu tFlowToIterate
 */
 
// tJava - Cititor streaming pentru fisiere uriase
java.io.BufferedReader reader = new java.io.BufferedReader(
    new java.io.FileReader(context.input_file),
    1024 * 1024  // buffer 1MB
);
 
String line;
long lineCount = 0;
while ((line = reader.readLine()) != null) {
    // Proceseaza linia
    lineCount++;
    if (lineCount % 100000 == 0) {
        System.out.println("Processed: " + lineCount);
        System.gc(); // Hint GC pentru job-uri cu rulare lunga
    }
}
reader.close();
 
// Fisier memory-mapped pentru acces aleator
java.io.RandomAccessFile raf = new java.io.RandomAccessFile(context.input_file, "r");
java.nio.channels.FileChannel channel = raf.getChannel();
java.nio.MappedByteBuffer buffer = channel.map(
    java.nio.channels.FileChannel.MapMode.READ_ONLY,
    0, channel.size()
);
// Proceseaza buffer-ul direct - foarte rapid pentru fisiere mari

// tFileOutputDelimited - Configurare performanta
 
/*
 * Settings:
 * - Append: false (mai rapid decat modul append)
 * - Include header: true (doar primul fisier in scenariu split)
 * - Compress output: true (pentru fisiere mari)
 *
 * Advanced:
 * - Flush buffer: 10000 (randuri inainte de flush)
 * - Create directory: true
 * - Split output: true (pentru output-uri foarte mari)
 * - Rows per split: 1000000
 */
 
// Writer buffered personalizat pentru throughput maxim
java.io.BufferedWriter writer = new java.io.BufferedWriter(
    new java.io.OutputStreamWriter(
        new java.io.FileOutputStream(context.output_file),
        "UTF-8"
    ),
    8 * 1024 * 1024  // buffer 8MB
);
 
// Scrie in batch-uri
StringBuilder batch = new StringBuilder();
int batchSize = 0;
int batchLimit = 10000;
 
// In bucla principala:
batch.append(row.field1).append(",")
     .append(row.field2).append("\n");
batchSize++;
 
if (batchSize >= batchLimit) {
    writer.write(batch.toString());
    batch.setLength(0);
    batchSize = 0;
}
 
// End - flush restul
if (batch.length() > 0) {
    writer.write(batch.toString());
}
writer.close();

// tJava - Setup monitorizare performanta
 
// La inceputul job-ului:
long jobStartTime = System.currentTimeMillis();
java.util.Map<String, Long> componentTimes = new java.util.HashMap<>();
java.util.Map<String, Long> componentRows = new java.util.HashMap<>();
globalMap.put("jobStartTime", jobStartTime);
globalMap.put("componentTimes", componentTimes);
globalMap.put("componentRows", componentRows);
 
Runtime runtime = Runtime.getRuntime();
System.out.println("=== Job Started ===");
System.out.println("Max Memory: " + runtime.maxMemory() / (1024*1024) + " MB");
System.out.println("Total Memory: " + runtime.totalMemory() / (1024*1024) + " MB");
System.out.println("Free Memory: " + runtime.freeMemory() / (1024*1024) + " MB");
 
// Dupa fiecare componenta majora (tJava):
long componentStart = System.currentTimeMillis();
globalMap.put("component_" + "tMap_1" + "_start", componentStart);
 
// La sfarsitul componentei:
long componentEnd = System.currentTimeMillis();
long componentStart = (Long) globalMap.get("component_" + "tMap_1" + "_start");
long duration = componentEnd - componentStart;
((java.util.Map<String, Long>) globalMap.get("componentTimes")).put("tMap_1", duration);
 
// La sfarsitul job-ului - sumar:
long jobEndTime = System.currentTimeMillis();
long jobStartTime = (Long) globalMap.get("jobStartTime");
System.out.println("\n=== Job Performance Summary ===");
System.out.println("Total Duration: " + (jobEndTime - jobStartTime) + " ms");
System.out.println("\nComponent Times:");
java.util.Map<String, Long> times =
    (java.util.Map<String, Long>) globalMap.get("componentTimes");
for (java.util.Map.Entry<String, Long> entry : times.entrySet()) {
    System.out.println("  " + entry.getKey() + ": " + entry.getValue() + " ms");
}

// tJava - Tracker utilizare memorie
 
public class MemoryMonitor {
    private static java.lang.management.MemoryMXBean memoryBean =
        java.lang.management.ManagementFactory.getMemoryMXBean();
 
    public static void logMemory(String checkpoint) {
        java.lang.management.MemoryUsage heapUsage = memoryBean.getHeapMemoryUsage();
 
        System.out.println(String.format(
            "[%s] Heap: Used=%dMB, Committed=%dMB, Max=%dMB",
            checkpoint,
            heapUsage.getUsed() / (1024*1024),
            heapUsage.getCommitted() / (1024*1024),
            heapUsage.getMax() / (1024*1024)
        ));
 
        // Verifica daca se apropie de limita de memorie
        double usedPercent = (double) heapUsage.getUsed() / heapUsage.getMax() * 100;
        if (usedPercent > 80) {
            System.out.println("WARNING: Memory usage at " + String.format("%.1f", usedPercent) + "%");
            System.gc(); // Sugereaza GC
        }
    }
}
 
// Apeleaza la checkpoint-uri:
MemoryMonitor.logMemory("After_tMap_1");
MemoryMonitor.logMemory("Before_BulkLoad");

#!/bin/bash
# Activeaza logging GC pentru job-urile Talend
 
export JAVA_OPTS="${JAVA_OPTS} \
    -Xlog:gc*:file=/var/log/talend/gc-%t.log:time,uptime,level,tags:filecount=5,filesize=100m \
    -XX:+PrintGCDetails \
    -XX:+PrintGCDateStamps \
    -XX:+PrintGCApplicationStoppedTime"
 
# Analizeaza log-urile GC cu GCViewer sau GCEasy
# Cauta:
# - Pauze GC lungi (>500ms)
# - Full GC frecvent
# - Scurgeri de memorie (heap creste in timp)

talend_optimization_checklist:
  memorie:
    - "Seteaza dimensiunea heap JVM corespunzatoare (Xms = Xmx pentru productie)"
    - "Foloseste G1GC sau ZGC pentru heap-uri mari"
    - "Activeaza string deduplication pentru job-uri cu mult text"
    - "Monitoreaza si logheaza utilizarea memoriei la checkpoint-uri"
    - "Curata obiectele mari dupa utilizare"
 
  paralelizare:
    - "Foloseste tParallelize pentru fluxuri de date independente"
    - "Partitioneaza datele pentru procesare paralela"
    - "Potriveste numarul de thread-uri cu core-urile CPU disponibile"
    - "Foloseste constructe thread-safe in contexte paralele"
    - "Sincronizeaza corect fluxurile paralele la reunire"
 
  baze_de_date:
    - "Foloseste bulk insert cu dimensiune batch corespunzatoare"
    - "Dezactiveaza indexii in timpul incarcarilor mari, reconstruieste dupa"
    - "Foloseste connection pooling pentru acces multi-threaded"
    - "Filtreaza datele la nivel SQL, nu in Talend"
    - "Foloseste prepared statements cu caching corespunzator"
 
  tMap:
    - "Alege modelul de lookup corespunzator in functie de dimensiunea datelor"
    - "Foloseste variabile pentru expresii repetate"
    - "Ordoneaza lookup-urile dupa selectivitate"
    - "Pre-filtreaza datele inainte de tMap-uri complexe"
    - "Considera tSortMergeJoin pentru join-uri foarte mari"
 
  fisiere:
    - "Foloseste I/O buffered cu dimensiuni mari de buffer"
    - "Comprima fisierele de output mari"
    - "Imparte output-urile foarte mari in fisiere multiple"
    - "Foloseste streaming pentru fisiere mai mari decat memoria"
    - "Proceseaza fisierele in chunk-uri cu tFlowToIterate"
 
  general:
    - "Profileaza job-urile pentru a identifica bottleneck-urile"
    - "Testeaza cu volume de date de productie"
    - "Documenteaza baseline-urile de performanta"
    - "Monitoreaza timpii de executie in TAC"
    - "Revizuieste periodic job-urile lente pentru optimizare"