Ottimizzazione join di intervalli

Un range join si verifica quando due relazioni vengono unite usando una condizione di appartenenza di un punto a un intervallo o una condizione di sovrapposizione di intervalli. L'uso dell'ottimizzazione dei join su intervalli in Databricks Runtime può migliorare notevolmente le prestazioni delle query.

In Databricks SQL, Azure Databricks ottimizza automaticamente i join di intervallo senza alcuna configurazione manuale. È anche possibile ottimizzare manualmente i join di intervalli usando hint di join o la configurazione della sessione per tutti i tipi di calcolo.

Join tra punti in intervallo

Un join di un punto nell'intervallo è un join la cui condizione contiene predicati che specifica che un valore di una relazione rientra tra due valori dell'altra relazione. Ad esempio:

-- using BETWEEN expressions
SELECT *
FROM points JOIN ranges ON points.p BETWEEN ranges.start and ranges.end;

-- using inequality expressions
SELECT *
FROM points JOIN ranges ON points.p >= ranges.start AND points.p < ranges.end;

-- with fixed length interval
SELECT *
FROM points JOIN ranges ON points.p >= ranges.start AND points.p < ranges.start + 100;

-- join two sets of point values within a fixed distance from each other
SELECT *
FROM points1 p1 JOIN points2 p2 ON p1.p >= p2.p - 10 AND p1.p <= p2.p + 10;

-- a range condition together with other join conditions
SELECT *
FROM points, ranges
WHERE points.symbol = ranges.symbol
  AND points.p >= ranges.start
  AND points.p < ranges.end;

Join di intervalli sovrapposti

Un intervallo di sovrapposizione è un join in cui la condizione contiene predicati che specificano una sovrapposizione di intervalli tra due valori da ciascuna relazione. Ad esempio:

-- overlap of [r1.start, r1.end] with [r2.start, r2.end]
SELECT *
FROM r1 JOIN r2 ON r1.start < r2.end AND r2.start < r1.end;

-- overlap of fixed length intervals
SELECT *
FROM r1 JOIN r2 ON r1.start < r2.start + 100 AND r2.start < r1.start + 100;

-- a range condition together with other join conditions
SELECT *
FROM r1 JOIN r2 ON r1.symbol = r2.symbol
  AND r1.start <= r2.end
  AND r1.end >= r2.start;

Ottimizzazione join di intervalli

L'ottimizzazione del join di intervallo viene eseguita per i join che:

• Avere una condizione che può essere interpretata come un punto in un intervallo o come un intervallo sovrapposto in un'intersezione di intervalli.
• Tutti i valori coinvolti nella condizione di join basata su un intervallo sono di tipo numerico (integrale, a virgola mobile, decimale), DATEo TIMESTAMP.
• Tutti i valori coinvolti nella condizione di join dell'intervallo sono dello stesso tipo. Nel caso del tipo decimale, anche i valori devono essere della stessa scala e precisione.
• Si tratta di un INNER JOIN, oppure, se si tratta di un punto nell'intervallo, di un LEFT OUTER JOIN con valore sul lato sinistro o di un RIGHT OUTER JOIN con valore sul lato destro.
• Disporre di una dimensione del contenitore, derivata automaticamente o manualmente.

Join con condizioni di uguaglianza numerica e di intervallo

Quando una condizione di join include sia una condizione di uguaglianza su una colonna numerica sia una condizione di intervallo, l'ottimizzatore può applicare il binning alla colonna numerica coinvolta nella condizione di uguaglianza, perché soddisfa i requisiti di tipo per l'ottimizzazione del join su intervalli. Ciò può comportare l'assegnazione della colonna di uguaglianza ai contenitori o l'esclusione dall'ottimizzazione, riducendo le prestazioni.

Per garantire che l'ottimizzazione del join di intervallo venga applicata solo alla condizione di intervallo desiderata, eseguire il cast delle colonne di uguaglianza numerica in STRING. Questo le esclude dalla valutazione come colonne di condizione dell'intervallo.

SELECT /*+ RANGE_JOIN(reference, 3306084) */
    reference.*, position.*
FROM position
INNER JOIN reference
    ON CAST(position.parent_index AS STRING) = CAST(reference.parent_index AS STRING)
    AND position.child_index BETWEEN reference.min_child_index AND reference.max_child_index;

Lo stesso modello si applica ad altre colonne numeriche usate come chiavi di uguaglianza, ad esempio DATE, identificatori integer o colonne di partizione cluster.

Dimensioni contenitore

Le dimensioni del bin sono un parametro di regolazione numerica che suddivide il dominio dei valori della condizione di gamma in più contenitori di dimensioni uguali. Ad esempio, con dimensioni bin pari a 10, l'ottimizzazione suddivide il dominio in contenitori con intervalli di lunghezza 10. Se si dispone di un punto nella condizione di intervallo di p BETWEEN start AND ende start è 8 e end è 22, questo intervallo di valori si sovrappone a tre bin di lunghezza 10: il primo contenitore da 0 a 10, il secondo contenitore da 10 a 20 e il terzo contenitore da 20 a 30. Solo i punti che rientrano negli stessi tre bin devono essere considerati come possibili corrispondenze di join per tale intervallo. Ad esempio, se p è 32, può essere escluso dall'intervallo tra start 8 e end 22, perché cade nell'intervallo da 30 a 40.

È possibile specificare la dimensione del contenitore usando un hint di join di intervallo nella query o impostando un parametro di configurazione della sessione. In Databricks SQL, la dimensione del bin viene derivata automaticamente quando è abilitata l'ottimizzazione automatica del join per intervallo.

Ottimizzazione automatica dei join su intervalli

In Databricks SQL Azure Databricks rileva automaticamente i join di intervallo idonei e deriva le dimensioni ottimali del bin eseguendo il campionamento della tabella intervalli. In questo modo viene rimossa la necessità di specificare manualmente le dimensioni di un contenitore tramite hint o la configurazione della sessione.

L'ottimizzazione automatica dei join di intervalli è abilitata per impostazione predefinita in Databricks SQL. Per disabilitarla, impostare la configurazione seguente:

SET spark.databricks.optimizer.autoRangeJoin.enabled = false;

Se si specifica una dimensione del bin tramite un hint di join di intervallo o una configurazione di sessione, tale valore ha la precedenza sulla dimensione del bin derivata automaticamente.

Abilitare il join di intervalli usando un suggerimento per il join di intervalli

Per abilitare l'ottimizzazione del range join in una query SQL, utilizzare un hint per il range join per specificare la dimensione del bin. L'hint deve contenere il nome della relazione di una delle relazioni unite e il parametro numeric bin size. Il nome della relazione può rappresentare una tabella, una vista o una sottoquery.

SELECT /*+ RANGE_JOIN(points, 10) */ *
FROM points JOIN ranges ON points.p >= ranges.start AND points.p < ranges.end;

SELECT /*+ RANGE_JOIN(r1, 0.1) */ *
FROM (SELECT * FROM ranges WHERE ranges.amount < 100) r1, ranges r2
WHERE r1.start < r2.start + 100 AND r2.start < r1.start + 100;

SELECT /*+ RANGE_JOIN(c, 500) */ *
FROM a
  JOIN b ON (a.b_key = b.id)
  JOIN c ON (a.ts BETWEEN c.start_time AND c.end_time)

#create minute table
minutes = spark.createDataFrame(
    [(0, 60), (60, 120)],
    "minute_start: int, minute_end: int"
)

#create events table
events = spark.createDataFrame(
    [(12, 33), (0, 120), (33, 72), (65, 178)],
    "event_start: int, event_end: int"
)

#Range_Join with "hint" on the from table
(events.hint("range_join", 60)
  .join(minutes,
    on=[events.event_start < minutes.minute_end,
    minutes.minute_start < events.event_end])
  .orderBy(events.event_start,
    events.event_end,
    minutes.minute_start)
  .show()
)

#Range_Join with "hint" on the join table
(events.join(minutes.hint("range_join", 60),
  on=[events.event_start < minutes.minute_end,
    minutes.minute_start < events.event_end])
  .orderBy(events.event_start,
    events.event_end,
    minutes.minute_start)
  .show()
)

È anche possibile inserire un hint di join di intervallo su uno dei DataFrame uniti. In tal caso, l'hint contiene solo il parametro di dimensione numerica del bin.

val df1 = spark.table("ranges").as("left")
val df2 = spark.table("ranges").as("right")

val joined = df1.hint("range_join", 10)
  .join(df2, $"left.type" === $"right.type" &&
     $"left.end" > $"right.start" &&
     $"left.start" < $"right.end")

val joined2 = df1
  .join(df2.hint("range_join", 0.5), $"left.type" === $"right.type" &&
     $"left.end" > $"right.start" &&
     $"left.start" < $"right.end")

Abilitare l'aggiunta a intervalli usando la configurazione della sessione

Se non si vuole modificare la query, specificare le dimensioni del bin come parametro di configurazione.

SET spark.databricks.optimizer.rangeJoin.binSize=5

Questo parametro di configurazione si applica a qualsiasi join con una condizione di intervallo. Tuttavia, una dimensione del contenitore diversa impostata tramite un suggerimento di join di intervallo esegue sempre l'override di quella impostata tramite il parametro.

Scegliere le dimensioni del contenitore

L'efficacia dell'ottimizzazione del join di intervallo dipende dalla scelta delle dimensioni del blocco appropriate.

Una piccola dimensione del contenitore comporta un numero maggiore di contenitori, che consente di filtrare le potenziali corrispondenze. Tuttavia, diventa inefficiente se la dimensione del bin è significativamente inferiore agli intervalli di valore rilevati e gli intervalli di valore si sovrappongono a più intervalli di bin. Ad esempio, con una condizione p BETWEEN start AND end, dove start è 1.000.000 e end è 1.999.999 e una dimensione del cestino pari a 10, l'intervallo di valori si sovrappone a 100.000 cestini.

Se la lunghezza dell'intervallo è abbastanza uniforme e nota, è consigliabile impostare le dimensioni del contenitore sulla lunghezza prevista tipica dell'intervallo di valori. Tuttavia, se la lunghezza dell'intervallo è variabile e asimmetrica, è necessario trovare un bilanciamento per impostare una dimensione del bin che filtra in modo efficiente gli intervalli brevi, impedendo al contempo la sovrapposizione di troppi bin sugli intervalli lunghi. Supponendo una tabella ranges, con intervalli tra le colonne start e end, è possibile determinare i diversi percentili del valore di lunghezza dell'intervallo asimmetrico utilizzando la query seguente.

SELECT
  map_from_arrays(
    ARRAY(0.5, 0.9, 0.99, 0.999, 0.9999),
    APPROX_PERCENTILE(
      end::DOUBLE - start::DOUBLE,
      ARRAY(0.5, 0.9, 0.99, 0.999, 0.9999)
    )
  ) AS bin_sizes
FROM
  ranges;

La conversione di ogni colonna in DOUBLE prima di sottrarre garantisce che la query funzioni sia che le colonne contengano valori numerici, DATE o TIMESTAMP.

Un'impostazione consigliata delle dimensioni del contenitore corrisponde al valore massimo al 90° percentile o al 99° percentile diviso per 10 oppure al valore al 99,9° percentile diviso per 100 e così via. La logica è:

• Se il valore in corrispondenza del 90° percentile è la dimensione del contenitore, solo il 10% delle lunghezze dell'intervallo di valore è maggiore dell'intervallo bin, quindi si estendono più di 2 intervalli bin adiacenti.
• Se il valore al 99° percentile corrisponde alla dimensione del bin, solo l'1% delle lunghezze dell'intervallo di valori si estende su più di 11 intervalli di bin consecutivi.
• Se il valore al 99,9° percentile è la dimensione del contenitore, solo lo 0,1% delle lunghezze dell'intervallo di valori si estende su più di 101 intervalli bin adiacenti.
• Lo stesso può essere ripetuto per i valori al 99,99°, al 99,999° percentile e così via, se necessario.

Il metodo descritto limita la quantità di intervalli di valori lunghi asimmetrici che si sovrappongono a più intervalli di bin. Il valore della dimensione del contenitore ottenuto in questo modo è solo un punto di partenza per l'ottimizzazione; i risultati effettivi possono dipendere dal carico di lavoro specifico.