Distributed Decision Trees

"I asked every worker not for its data but for a tally. Sixteen numbers came back from each, we added them up, and the whole forest agreed on where to cut."

A Split Finder Who Never Saw a Single Row

The previous sections distributed models with smooth, additive objectives: linear and logistic regression, then support vector machines, where the gradient or the dual update decomposes cleanly across data shards. A decision tree is a different animal. It has no global parameter vector and no gradient of a single loss; it is grown greedily, one node at a time, by a discrete search for the split that best separates the data reaching that node. That discreteness is exactly what makes the distributed version interesting. We cannot all-reduce a gradient, because there is no gradient. What we can all-reduce is the set of sufficient statistics the split search consumes, and the art of the chapter is recognizing that those statistics are small, additive, and therefore cheap to combine.

1. The Split-Finding Bottleneck Beginner

Growing a tree is a recursion. At the root, all $N$ training examples are present; the algorithm picks one feature and one threshold that splits them into a left child and a right child so that each child is purer (more dominated by one label, or lower in variance) than the parent. It then recurses on each child with the examples that fell into it, stopping when a node is pure enough, small enough, or deep enough. Everything expensive about this recursion lives in one step, repeated at every node: find the best split.

To score a candidate split of a node on a feature, you need to know, for the examples at that node, how the label statistics divide between the two sides of the threshold. For classification that means counts of each class on the left and on the right; for regression it means the sum and sum of squares of the targets. The standard quality measure is the impurity reduction, often called the gain. Writing $\text{imp}(\cdot)$ for a node's impurity (Gini or entropy for classification, variance for regression), a split of a parent node into a left child $L$ and a right child $R$ has gain

where $n_L$ and $n_R$ are the numbers of examples falling left and right, and $n = n_L + n_R$. The greedy tree picks, over all features and all thresholds, the split that maximizes this gain. Notice what the formula actually consumes: only counts and label sums on each side of the threshold. It never needs the rows themselves, only their tallies. That observation is the seed of the entire distributed construction.

On a single machine the classic way to evaluate every threshold for a feature is to sort the node's examples by that feature and sweep, maintaining running left and right statistics. That is $O(N \log N)$ per feature per node, and worse, it assumes the rows are all in one place. When the $N$ examples are sharded across $K$ machines, no machine can sort the global order, and shipping the rows to one machine to sort them would move exactly the terabytes we sharded to avoid. The bottleneck is not the arithmetic of the gain; it is gathering the statistics the gain needs from data that lives everywhere at once.

2. The Histogram Method Intermediate

The histogram method, the idea behind every scalable tree library, removes the sort entirely and replaces continuous thresholds with a fixed, small set of candidate cut points. Before training, each feature's range is divided into $B$ bins (typically 64 or 255), and every feature value is replaced by its bin index. A node's statistics for one feature are then a histogram with $B$ buckets: for classification, each bucket holds the count of examples and the count of positives that landed in that bin; for boosting, each bucket holds the sum of gradients and the sum of Hessians. Evaluating splits for that feature becomes a single left-to-right sweep over $B$ buckets, accumulating a running prefix on the left and reading the complement on the right, which costs $O(B)$ rather than $O(N \log N)$.

The distributed payoff is immediate. A histogram is additive across data shards. If worker $k$ builds the histogram of its own rows for a feature, then the global histogram for that feature is the bucket-wise sum of the $K$ local histograms. Summing one vector held on each worker and sharing the result is precisely the all-reduce collective introduced in Section 4.3; here it carries histogram buckets instead of gradient components. After the all-reduce every worker holds the identical merged histogram, runs the identical $O(B)$ sweep, and selects the identical best split. No worker ever saw another worker's rows; they exchanged only tallies. Figure 12.3.1 traces the flow for one feature.

The cost accounting is the whole point. Each worker sends, per node, one histogram per feature: $F$ features times $B$ bins times the bytes of the per-bucket statistics. For 200 features, 255 bins, and two 4-byte statistics per bucket, that is about 400 kilobytes per node, regardless of whether the shard holds ten thousand rows or ten billion. The all-reduce cost depends on $B$, $F$, and the number of workers, but not on $N$. Doubling the data per worker doubles the local binning work, which is embarrassingly parallel, and changes the communication not at all. This is why the histogram tree is the scalable tree: it converts a data-movement problem whose size is the dataset into a communication problem whose size is a fixed summary.

3. A Sharded Split That Matches the Whole-Data Split Intermediate

The claim worth verifying is exactness. Because histogram addition is just integer and floating-point summation, the merged histogram is bit-for-bit the histogram you would have built on the pooled data, so the split chosen from it is the split a single machine would have chosen. The code below grows the work of one node. It bins a single feature, builds the full-data histogram as a reference, then shards the rows across $K$ workers, has each worker build a local histogram, all-reduces them by summation, and finds the best Gini-gain threshold from the merged result. It then checks that the merged histogram equals the full one and that the two splits agree.

import numpy as np

rng = np.random.default_rng(7)
N, K, B = 60_000, 6, 16            # examples, workers (row shards), feature bins
# One feature, binned to B buckets; binary label. Find the best threshold split.
feat_bin = rng.integers(0, B, size=N)            # pre-quantized feature value in [0, B)
# Label probability rises with the bin index, so a real split exists.
p = 0.15 + 0.70 * (feat_bin / (B - 1))
y = (rng.random(N) < p).astype(np.int64)

def local_histogram(bins, labels, B):
    """Per-bin (count, positive-count) statistics over THIS worker's shard."""
    cnt = np.bincount(bins, minlength=B).astype(np.float64)
    pos = np.bincount(bins, weights=labels, minlength=B).astype(np.float64)
    return np.stack([cnt, pos])               # shape (2, B): row 0 count, row 1 positives

def gini(pos, cnt):
    """Gini impurity of a node with `pos` positives out of `cnt` examples."""
    if cnt == 0:
        return 0.0
    p1 = pos / cnt
    return 1.0 - p1 * p1 - (1.0 - p1) ** 2

def best_split_from_hist(hist):
    """Scan thresholds t: left = bins < t, right = bins >= t. Maximize Gini gain."""
    cnt, pos = hist[0], hist[1]
    total_cnt, total_pos = cnt.sum(), pos.sum()
    parent = gini(total_pos, total_cnt)
    best_gain, best_t = -1.0, None
    cum_cnt = np.cumsum(cnt)                   # examples with bin < t, for t = 1..B-1
    cum_pos = np.cumsum(pos)
    for t in range(1, len(cnt)):
        lc, lp = cum_cnt[t - 1], cum_pos[t - 1]
        rc, rp = total_cnt - lc, total_pos - lp
        if lc == 0 or rc == 0:
            continue
        child = (lc / total_cnt) * gini(lp, lc) + (rc / total_cnt) * gini(rp, rc)
        gain = parent - child
        if gain > best_gain:
            best_gain, best_t = gain, t
    return best_t, best_gain

# --- Single-machine reference: one histogram over ALL data, then split. ---
full_hist = local_histogram(feat_bin, y, B)
t_full, g_full = best_split_from_hist(full_hist)

# --- Distributed: shard rows, each worker builds a LOCAL histogram. ---
shards = np.array_split(np.arange(N), K)
local_hists = [local_histogram(feat_bin[s], y[s], B) for s in shards]   # K workers

# All-reduce: sum the (2, B) histograms element-wise across workers.
merged_hist = np.sum(local_hists, axis=0)
t_dist, g_dist = best_split_from_hist(merged_hist)

bytes_per_worker = local_hists[0].size * 8       # one shard's histogram payload
print("workers (row shards)      :", K)
print("feature bins B            :", B)
print("histogram floats / worker :", local_hists[0].size, f"({bytes_per_worker} bytes)")
print("full-data best threshold  :", t_full, " gain =", f"{g_full:.6f}")
print("all-reduce best threshold :", t_dist, " gain =", f"{g_dist:.6f}")
print("merged == full histogram  :", bool(np.array_equal(merged_hist, full_hist)))
print("splits identical          :", t_full == t_dist and abs(g_full - g_dist) < 1e-15)

workers (row shards)      : 6
feature bins B            : 16
histogram floats / worker : 32 (256 bytes)
full-data best threshold  : 8  gain = 0.069405
all-reduce best threshold : 8  gain = 0.069405
merged == full histogram  : True
splits identical          : True

The split found over six blind shards is the split found over the pooled data, down to the last digit of the gain, and the histogram each worker sent was 256 bytes whether its shard held ten thousand rows or ten billion. To grow a full tree you repeat Code 12.3.1 at every node, building one histogram per feature and all-reducing the stack; a useful efficiency trick, the histogram subtraction property, lets a child reuse its parent's histogram minus its sibling's so only the smaller child is ever scanned. The mechanism, though, is exactly what you just ran: summarize locally, all-reduce, decide identically everywhere.

4. Data-Parallel Versus Feature-Parallel Trees Advanced

Code 12.3.1 shards the rows: every worker holds all features for a slice of the examples and builds the full set of feature histograms over its slice. This is data-parallel tree building, and it is the default for the common case where rows vastly outnumber features. Its communication per node is the all-reduce of $F$ histograms of $B$ buckets, so $O(F \cdot B)$ traffic, and its local work is binning the shard, which parallelizes perfectly in $N$. The weakness shows when $F \cdot B$ itself is large, a wide dataset with thousands of features, because then the histogram payload, repeated at every node, becomes the bottleneck even though it is independent of $N$.

The alternative is to shard the columns. In feature-parallel tree building, each worker holds all rows but only a subset of the features. Each worker finds the best split among its own features locally, with no histogram exchange at all, and the workers then exchange only the single best candidate each found, a few numbers, to elect the global winner. Communication per node is tiny, $O(K)$ candidates rather than $O(F \cdot B)$ buckets. The catch is twofold: every worker must hold every row, so the data cannot exceed one machine's capacity in the row dimension, and once the winning split is chosen, the partition of rows into left and right child must be broadcast to all workers so they can re-segment their columns, which costs $O(N)$ communication per node and grows with the data. Table 12.3.1 lays the two strategies side by side.

Production systems mix the two. LightGBM offers both a data-parallel and a feature-parallel mode and a voting-parallel variant that all-reduces only the top-$k$ candidate features per worker to shrink the $O(F \cdot B)$ payload; XGBoost's distributed mode is histogram all-reduce over row shards. The decision is the recurring one from Chapter 3: estimate the communication volume each strategy implies for your $N$, $F$, $B$, and $K$, and pick the one whose dominant term is smallest. There is no universally best partition, only the one matched to the shape of your data.

5. Choosing the Bin Boundaries with Quantile Sketches Advanced

One question was assumed away above: where do the bin edges come from? For the histogram method to be both accurate and balanced, each feature's bins should carry roughly equal numbers of examples, which means the cut points should sit at the quantiles of that feature's distribution, the median, the quartiles, and so on for $B$ buckets. Computing exact quantiles requires sorting the feature globally, the very $O(N \log N)$ all-to-one operation we sharded to avoid. The way out is the same way out as everywhere else in this book: an approximate, mergeable summary.

A quantile sketch is a small, fixed-size data structure that summarizes the distribution of a stream of values and answers "what value sits at quantile $q$?" to within a bounded error, using memory that does not grow with the number of values seen. Crucially, sketches are mergeable: the sketch of a shard plus the sketch of another shard combine into a sketch of the union, with the same error guarantee. So each worker builds a sketch of its own feature values, the sketches are merged across workers (one more all-reduce-shaped combine), and the merged sketch yields the global bin boundaries that every worker then uses to bin its rows. These structures, the Greenwald-Khanna sketch and the t-digest among them, were introduced as a streaming and MapReduce tool in Section 6.8; distributed tree building is one of their most important consumers, because XGBoost's weighted quantile sketch is exactly how it proposes candidate splits over sharded data.

import xgboost as xgb
from dask.distributed import Client
from xgboost import dask as dxgb

client = Client("scheduler-address:8786")     # K workers, data already sharded across them
dtrain = dxgb.DaskDMatrix(client, X_sharded, y_sharded)   # rows stay on their workers

out = dxgb.train(
    client,
    {"tree_method": "hist", "max_bin": 255, "objective": "binary:logistic"},
    dtrain, num_boost_round=300,
)                                              # histogram all-reduce + quantile sketch, internal
model = out["booster"]

6. From One Tree to the Forest Beginner

Everything in this section grew a single tree over sharded data. Two large structures are built on exactly this foundation. The first is the random forest, many independent trees each grown on a bootstrap sample of rows and a random subset of features; because the trees are independent, they parallelize along a second axis (one tree per worker group, or even one tree per worker), and Section 12.4 takes up that ensemble-level parallelism and how it composes with the node-level histogram all-reduce you just built. The second is gradient boosting, where trees are grown in sequence, each correcting the errors of those before it, so the per-node histogram, now of gradients and Hessians rather than class counts, is all-reduced for every node of every tree in a long dependent chain. That dependent chain is what makes boosting the harder and more interesting distributed problem, and Section 12.5 develops it as the production workhorse of tabular machine learning. The histogram all-reduce of Code 12.3.1 is the atom from which both are assembled.

The distributed decision tree is, in the end, a small idea with large consequences. A tree needs only tallies to choose its splits; tallies add; addition across machines is an all-reduce; and the all-reduce of a fixed-size summary costs the same whether the data is gigabytes or petabytes. Hold that atom firmly, because Section 12.4 replicates it across an ensemble and Section 12.5 chains it through boosting, and both are just this section's histogram all-reduce, scaled out.