Spark batch LLM feldolgozás

Az adatszuverenitás és a shadow AI

A szervezetekben ma egyre nagyobb nyomás nehezedik az alkalmazottakra, hogy AI-t használjanak a napi munkában. Ahol a szervezet nem biztosít megbízható, vállalati keretek között működő AI eszközt, ott az emberek saját megoldásokat keresnek – és érzékeny vállalati adatokat másolnak külső felhőszolgáltatásokba. Ez a shadow AI jelensége: nem rosszindulatból, hanem kényszerűségből.

A platform Spark transformer rétege erre ad választ: az AI feldolgozás ott fut, ahol az adat van. Az adat nem másolódik ki, nem kerül fel felhőszolgáltatásba, nem hagyja el a szervezet infrastruktúráját. A vállalat tudásának feldolgozása, gazdagítása és lekérdezése teljes egészében on-premise történik – a chatbot csak az egyik lehetséges belépési felület felette.

Mire való?

Az egymás utáni, szekvenciális LLM hívások sok dokumentum vagy sor feldolgozásakor szűk keresztmetszetet alkotnak. A Spark transformer réteg ezt oldja fel: a sorok párhuzamosan, konfigurálható batch mérettel kapnak LLM hívást.

Tipikus felhasználási területek:

• Vállalati adatok AI-alapú gazdagítása: meglévő adatbázis tartalmának LLM-mel történő kiegészítése – osztályozás, összefoglalás, struktúrálás, fordítás – anélkül, hogy az adat elhagyná a rendszert.
• Ingestion pipeline: dokumentumok összefoglalása, kulcsszó-kinyerés, NER annotáció feltöltés előtt.
• Batch enrichment: ütemezett feldolgozás nagy rekordszámon – az agent futhat chatbot kérés nélkül is, önálló batch jobként.

Az osztályhierarchia

Transformer  (SparkML alap)
  └── BaseTransformer
        ├── BaseLLMTransformer              ← párhuzamos LLM hívások – ezt kell leszármaztatni
        │     ├── SummaryToolTransformer    ← platform komponens (összefoglalás)
        │     └── BatchedSummaryTransformer ← platform komponens (összefoglalás + kulcsszavak)
        └── BaseBatchedTransformer          ← nem-LLM batch feldolgozás
              └── AbstractIngestionTransformer  ← dokumentum-alapú ingestion – ld. lentebb

A SummaryToolTransformer és a BatchedSummaryTransformer a platform ingestion pipeline-jának belső komponensei – névvel hivatkozhatók, de plugin fejlesztők nem leszármaztatják őket. Az LLM-alapú saját transformer mindig a BaseLLMTransformer-ből indul; a nem-LLM, dokumentum-beszerző/-tároló transformer az AbstractIngestionTransformer-ből (részletek: BaseBatchedTransformer és AbstractIngestionTransformer).

BaseLLMTransformer – az alap

Konstruktorban a system prompt és az opcionális JSON response format adható meg. A konfigurálható paraméterek builder-stílusú setter-ekkel láncolhatók:

Három kötelező abstract metódus

// Melyik sorokat kell feldolgozni?
// false → kihagyja (pl. ha a kimeneti oszlop már tartalmaz értéket → idempotencia)
protected abstract boolean isNeedTransform(Row r);

// Dataset előkészítése: kimeneti oszlopok felvétele null értékkel
protected abstract Dataset<Row> prepareDataSet(Dataset<?> dataset);

// Az LLM válasz feldolgozása → Map<oszlopnév, érték>
// A map kulcsai az output oszlopok nevei, értékei a beírandó adatok
protected abstract Map<String, Object> parseResponseJson(Row row, String retJson)
    throws Exception;

Teljes példa: GroundTrueQuestionGeneratorTransformer

A következő valódi implementáció RAG dokumentumokból generál ground truth kérdés-válasz párokat IR metrikák méréséhez. Jól mutatja a saját transformer tipikus felépítését: strukturált JSON kimenet, több output oszlop, és domain-specifikus minőségi szabályok a promptban.

Az eredmény két oszlopba kerül: a kérdések (queries) és a dokumentum–kérdés megfeleltetések (qrels) – a standard BEIR/TREC formátumnak megfelelően.

public class GroundTrueQuestionGeneratorTransformer extends BaseLLMTransformer {

    // ── Kimeneti sémák ──────────────────────────────────────────────────────────
    private static final ArrayType QUERY_RETURN_TYPE =
        DataTypes.createArrayType(DataTypes.createStructType(new StructField[]{
            new StructField("id",           DataTypes.StringType, true, Metadata.empty()),
            new StructField("query",        DataTypes.StringType, true, Metadata.empty()),
            new StructField("type",         DataTypes.StringType, true, Metadata.empty()),
            new StructField("answer",       DataTypes.StringType, true, Metadata.empty()),
            new StructField("importantWords",
                DataTypes.createArrayType(DataTypes.StringType), true, Metadata.empty()),
    }));

    private static final ArrayType QREL_RETURN_TYPE =
        DataTypes.createArrayType(DataTypes.createStructType(new StructField[]{
            new StructField("query_id",  DataTypes.StringType, true, Metadata.empty()),
            new StructField("doc_id",    DataTypes.StringType, true, Metadata.empty()),
            new StructField("text",      DataTypes.StringType, true, Metadata.empty()),
            new StructField("relevance", DataTypes.FloatType,  true, Metadata.empty()),
    }));

    private String queriesCol;
    private String qrelsCol;

    // ── System prompt – minőségi szabályok a kérdésalkotáshoz ──────────────────
    private static final String SYSTEM_PROMPT = """
            Szerepköröd: RAG rendszer IR metrikáinak méréséhez ground truth kérdésgenerátor.
            FONTOS SZABÁLYOK:
              - A kérdés minősége fontosabb, mint a mennyiség. Ha nem alkotható minőségi
                kérdés, adj vissza ÜRES JSON tömböt.
              - Csak olyan kérdést fogalmazz meg, amire van válasz a szövegben.
              - Utaló szavakat ("ilyen", "olyan", "ez", "az") tartalmazó kérdések tiltottak.
              - Semmitmondó kérdések tiltottak.
            FELADAT:
              - Válassz ki 2-5 releváns mondatot válaszként (answer).
              - Fogalmazz meg rá kérdést, amelyre ez a válasz tökéletesen illeszkedik.
              - Ha lehetséges, különböző típusú kérdéseket alkotj:
                factoid | entity_lookup | definition | reasoning | contextual
            """;

    // ── JSON response format (constrained decoding) ─────────────────────────────
    private static final String RESPONSE_FORMAT = """
            {
              "type": "json_schema",
              "json_schema": {
                "name": "ground_truth_output",
                "schema": {
                  "type": "object",
                  "properties": {
                    "resp": {
                      "type": "array",
                      "minItems": 0,
                      "maxItems": 5,
                      "items": {
                        "type": "object",
                        "additionalProperties": false,
                        "properties": {
                          "question":      { "type": "string" },
                          "answer":        { "type": "string" },
                          "type":          { "type": "string",
                                             "description": "factoid|entity_lookup|definition|reasoning|contextual" },
                          "importantWords":{ "type": "array",
                                             "items": { "type": "string" } }
                        },
                        "required": ["question", "answer", "type", "importantWords"]
                      }
                    }
                  },
                  "required": ["resp"]
                }
              }
            }
            """;

    public GroundTrueQuestionGeneratorTransformer() throws Exception {
        super();
        setSystemPrompt(SYSTEM_PROMPT, RESPONSE_FORMAT);
        setRetryPolicy(new RetryPolicy(0, 2_000, 10_000));
    }

    public GroundTrueQuestionGeneratorTransformer setOutputCols(
            String queriesCol, String qrelsCol) {
        this.queriesCol = queriesCol;
        this.qrelsCol   = qrelsCol;
        return this;
    }

    @Override public String uid() { return "GroundTrueQuestionGeneratorTransformer"; }

    // ── Séma kiterjesztés – két új Array oszlop ─────────────────────────────────
    @Override
    public StructType transformSchema(StructType schema) {
        StructType s = schema;
        if (!hasField(s, queriesCol)) s = s.add(queriesCol, QUERY_RETURN_TYPE, true);
        if (!hasField(s, qrelsCol))   s = s.add(qrelsCol,   QREL_RETURN_TYPE,  true);
        return s;
    }

    // ── Minden sort feldolgozunk (nincs idempotencia-szűrés) ────────────────────
    @Override
    protected boolean isNeedTransform(Row r) { return true; }

    @Override
    protected Dataset<Row> prepareDataSet(Dataset<?> dataset) {
        return dataset
            .withColumn(queriesCol, lit(null).cast(QUERY_RETURN_TYPE))
            .withColumn(qrelsCol,   lit(null).cast(QREL_RETURN_TYPE));
    }

    // ── LLM válasz feldolgozása → queries + qrels sorok ────────────────────────
    @SuppressWarnings({"rawtypes", "unchecked"})
    @Override
    protected Map<String, Object> parseResponseJson(Row row, String retJson)
            throws Exception {
        Map<String, Object> valueMap =
            objectMapper.readValue(retJson, new TypeReference<>(){});
        List<Map> resultList = (List) valueMap.get("resp");

        List<Row> queryRows = new ArrayList<>();
        List<Row> qrelRows  = new ArrayList<>();
        String docId = row.getAs("_key");

        for (Map result : resultList) {
            String id       = UUID.randomUUID().toString();
            String question = (String) result.get("question");
            String type     = (String) result.get("type");
            String answer   = (String) result.get("answer");

            // Spark Scala-kompatibilis tömb az importantWords-höz
            Object rawWords = result.get("importantWords");
            WrappedArray<String> words = rawWords instanceof Collection
                ? WrappedArray$.MODULE$.make(((Collection) rawWords).toArray())
                : null;

            queryRows.add(RowFactory.create(id, question, type, answer, words));
            qrelRows .add(RowFactory.create(id, docId, row.getAs(inputCol), 1f));
        }

        // Spark Scala-kompatibilis tömb a sorokhoz
        return Map.of(
            queriesCol, WrappedArray$.MODULE$.make(queryRows.toArray()),
            qrelsCol,   WrappedArray$.MODULE$.make(qrelRows.toArray())
        );
    }
}

Használat

// RAG dokumentumok betöltése (pl. egy domain teljes PageEmbedding gyűjteménye)
Dataset<Row> pages = loadPagesFromSource();

GroundTrueQuestionGeneratorTransformer generator =
    new GroundTrueQuestionGeneratorTransformer()
        .setOutputCols("queries", "qrels")
        .setBatchSize(5);

Dataset<Row> enriched = generator.transform(pages);

// Kérdések és qrel-ek kinyerése a kiértékelő pipeline-hoz
enriched.selectExpr("explode(queries) as q").select("q.*").show();
enriched.selectExpr("explode(qrels) as r").select("r.*").show();

BaseBatchedTransformer és AbstractIngestionTransformer – dokumentum-alapú ingestion

A BaseBatchedTransformer a BaseLLMTransformer nem-LLM testvére: ugyanazt a párhuzamos, batch-elt sor-feldolgozó keretet adja (repartition(1) + ThreadPoolExecutor, ugyanaz a három abstract metódus – isNeedTransform, prepareDataSet, processRow), de LLM hívás nélkül – tetszőleges soronkénti mellékhatásra (fájlírás, hálózati hívás, formátumkonverzió) használható.

Az AbstractIngestionTransformer erre épül: egy kész, dokumentum-alapú ingestion transformer, amely DocumentMetaData sorokból álló Dataset-et fogad, és a platform meglévő ingestion staging formátumába (source.json + tartalom fájlok) írja – ugyanabba a formátumba, amit a manuális dokumentum-feltöltés és a platform saját scraper rétege is használ. A kimenet így minden további teendő nélkül bekerül a normál embedding/lemmatizálási pipeline-ba, amint a batch a feldolgozó sorba kerül.

DocumentMetaData – a bemeneti sor

Feldolgozási logika soronként

• Ha text ki van töltve → változatlanul tárolja txt formátumban.
• Ha csak url → letölti a tartalmat, és a detektált tartalomtípus alapján dönt:
  • .doc/.docx → PDF-re konvertálja (a platform meglévő DocumentLoader.convertToPDF metódusával), és PDF-ként tárolja – a lapokra bontást, OCR-t és mondatgenerálást a downstream pipeline végzi el, ugyanúgy, mint bármely feltöltött PDF esetén.
  • már PDF → változatlanul tárolja.
  • text/html → a látható szöveget kinyeri (Jsoup), txt-ként tárolja.
  • egyéb → nyers szövegként tárolja.

Resume-safe URL deduplikáció

Egy batch könyvtár csak akkor törlődik, ha a benne lévő tartalom sikeresen eljutott az adatbázisig (downstream feldolgozás). Ha egy scrape félbeszakad, a már letöltött, de még fel nem dolgozott batch könyvtárak a helyükön maradnak. A transformer ezt kihasználja: prepareDataSet()-ben, egyszer a batch indulásakor, összegyűjti az adott domain többi (nem a jelenleg induló) batch könyvtárában található fájlokat egy memóriabeli fájlnév-indexbe. Soronként, letöltés előtt megnézi, szerepel-e ott már az adott URL-hez tartozó fájl ({sha256(url)}.{format} néven) – ha igen, onnan másolja át a tartalmat, HTTP letöltés és doc→PDF konverzió nélkül.

Használat

// 1. Saját, domain-specifikus leszármazott – nincs felülírandó metódus és
//    nincs saját konstruktor sem (l. fenti figyelmeztetés)
public class MyIngestionTransformer extends AbstractIngestionTransformer {
}

// 2. DocumentMetaData sorok összeállítása – a plugin saját beszerzési logikájával
//    (pl. egy katalógusoldal bejárásával, majd a talált dokumentum-URL-ek kigyűjtésével)
List<DocumentMetaData> rows = new ArrayList<>();
DocumentMetaData doc = new DocumentMetaData();
doc.setName("Példa dokumentum");
doc.setUrl("https://forras.example.com/dokumentum.docx");
doc.setSystem("MY_SYSTEM");
rows.add(doc);

// 3. Dataset építése és a transformer futtatása
Dataset<DocumentMetaData> dataset =
    sparkSessionStorage.getSession().createDataset(rows, Encoders.bean(DocumentMetaData.class));

String batchId = UUID.randomUUID().toString();
MyIngestionTransformer transformer = (MyIngestionTransformer) new MyIngestionTransformer()
    .setDomain("MY_DOMAIN")
    .setSystem("MY_SYSTEM")
    .setBatchId(batchId);

Dataset<Row> result = transformer.transform(dataset.toDF());

// 4. Beütemezés a feldolgozó sorba – az injektált IngestService platform service
//    végzi: kikényszeríti a Spark lusta kiértékelést, BatchStatus-t ment Redis-be,
//    majd a batchId-t a feldolgozó sorba lökve elindítja a downstream pipeline-t
ingestService.startIngest(batchId, result);

IngestService – beütemezés és állapotkövetés

Az IngestService Spring-managed platform service (egyszerű @Autowired-del injektálható), amely a transformer lefutása után kapott Dataset<Row>-ot fogadja és intézi a beütemezést:

• startIngest(String batchId, Dataset<Row> rows) – kikényszeríti a Spark lusta kiértékelést (rows.count()), BatchStatus-t ír Redis-be (accepted darabszám, createdAt, status: QUEUED ha van feldolgozandó sor, DONE ha nulla), majd a batchId-t a feldolgozó sorba (rag.redis.ingest-queue) löki. Visszatérési érték: a mentett BatchStatus.
• getIngestStatus(String batchId) – visszaadja a Redis-ben tárolt BatchStatus-t (pl. egy admin tool státusz-lekérdezéséhez).

BatchStatus mezői: createdAt, status (String), accepted (Long), errors (List<DocumentError>), finishedAt. DocumentError mezői: system, id, errorMessage – egy adott dokumentum feldolgozási hibájának leírására (jelenleg az IngestService maga nem tölti fel, a hívó oldal saját hibakezeléséhez áll rendelkezésre).

Chatbot-független batch feldolgozás

A transformer réteg nem kötődik a chatbot pipeline-hoz. Önálló Java/Spark programból, adminisztrációs tool-ból vagy ütemezett batchből is futtatható, ha a Spark session és a platform LLM kliensek elérhetők.

Tipikus önálló feldolgozási minta:

// 1. Adatforrás betöltése (pl. CSV, Parquet, JDBC)
Dataset<Row> rawData = spark.read()
    .option("header", "true")
    .csv("/data/input/records.csv");

// 2. Transformer alkalmazása
SummaryToolTransformer summarizer = new SummaryToolTransformer(null);
summarizer.setOutputCols("summary").setBatchSize(10);
Dataset<Row> enriched = summarizer.transform(rawData);

// 3. Eredmény visszaírása
enriched.write()
    .mode(SaveMode.Overwrite)
    .parquet("/data/output/enriched_records");

// 4. Vagy visszaírható a vállalati rendszerbe (JDBC, API, stb.)