Nodo: join

Accoppia due flussi per chiave, come un hash join: il lato build si legge per intero e si tiene in memoria, il lato probe scorre in streaming e cerca il suo corrispondente. È uno dei due modi di far confluire flussi nel grafo; l'altro è il union.

0/5

Scegli una scena e premi ▶ per guardarla scorrere, oppure → per andare avanti un passo alla volta.

Metodo	.join(id, { build, probe, how, buildKey, probeKey, merge, onDuplicateBuildKey?, bufferProbe? }, { deadLetter?, purpose?, writeReport? })
Ingressi	2: uno sulla porta build, uno sulla porta probe
Uscite	≥1 (fan-out libero)
Dead-letter	sì, solo per gli errori lato probe
Codice	src/core/join.ts

Quando usarlo

Quando il lato di destra (il build) è piccolo abbastanza da stare in memoria: anagrafiche, listini, tabelle di decodifica. Se è grande e serve solo per poche chiavi, è un lookup. Se i due flussi hanno la stessa forma e vanno solo messi in fila, è un union.

La configurazione

Campo	Default	Effetto
build	—	Id del nodo che alimenta il lato in memoria
probe	—	Id del nodo che alimenta il lato in streaming
how	—	"inner": scarta i probe senza corrispondenza. "left": li tiene, merge riceve undefined
buildKey, probeKey	—	Chiave (stringa) dei due lati
merge	—	(build | undefined, probe) => output
onDuplicateBuildKey	"error"	"error": una chiave ripetuta nel build ferma la run. "last-wins": vince l'ultima
bufferProbe	false	Legge tutto il probe in memoria: serve quando i due lati discendono dallo stesso nodo

Esempio

graph("fatture-arricchite")
  .source("fatture", sources.fatture)
  .source("clienti", sources.clienti)
  .transform("fatture", "fattura", validate("fattura", fatturaSchema))
  .transform("clienti", "cliente", validate("cliente", clienteSchema))
  .join("arricchisci", {
    build: "cliente", // piccolo: in memoria
    probe: "fattura", // grande: in streaming
    how: "left", // una fattura senza cliente è un caso valido
    buildKey: (c: Cliente) => c.id,
    probeKey: (f: Fattura) => f.clienteId,
    merge: (c, f) => ({ ...f, ragioneSociale: c?.nome ?? null }),
  })
  .sink("arricchisci", fatti)
  .build();

Build e probe dallo stesso nodo

Il pattern «aggrega e ricongiungi» — una source, un groupBy come build e la source stessa come probe — si bloccherebbe: il fan-out a monte non lascia avanzare il probe finché il build non è letto tutto. build() lo rifiuta, a meno di dichiarare bufferProbe: true, che legge tutto il probe in memoria.

.transform("righe", "totali", groupBy(/* … */))          // build: un totale per cliente
.transform("righe", "dettaglio", mapTransform(/* … */))  // probe: ogni riga
.join("con-totale", { build: "totali", probe: "dettaglio", how: "inner", /* … */, bufferProbe: true })

Alternative senza memoria extra: due source separate, o un lookup.

Errori

Caso	Esito
errore lato probe (chiave, merge) con deadLetter	il record probe esce come DeadLetter
errore lato probe senza deadLetter	ramo catch sotto un try-catch, altrimenti fail-fast
chiave duplicata nel build con onDuplicateBuildKey: "error"	la run si ferma sempre: non è colpa di un record probe

Dal vivo

Mentre il build si materializza, il join legge il probe in anticipo e lo tiene da parte: fino a 64 record, o tutto il probe con bufferProbe: true. Quei record sono il suo in lavorazione nei contatori dal vivo, e scendono a zero man mano che il join li confronta. Un numero alto e fermo vuol dire che il lato build è lento (convenzione del report).

Anti-pattern

• Un join con un build che non sta in memoria: non esistono join a finestra. Si riduce il build con un'aggregazione a monte, o si usa un lookup.
• Un join per unire flussi della stessa forma: è un union.