Spark Triangle Count, in depth: wedges, degree ordering, GraphX and GraphFrames, and a skew-aware DataFrame implementation

What is being counted

Treat the graph as undirected and simple: no self-loops and at most one edge between any pair. For a vertex v, the local triangle count t(v) is the number of triangles that contain v. The global count T is the number of distinct triangles. Every triangle contains three vertices, so the sum of t(v) over all vertices equals 3T.

The local clustering coefficient of v is t(v) divided by the number of pairs of neighbours, d(v)(d(v) - 1) / 2, where d(v) is the degree. It is the fraction of v's neighbour pairs that are themselves connected. A value near 1 means v sits inside a dense cluster; a value near 0 for a high-degree vertex means v connects many people who do not know each other, which is the shape of a hub, a broker or a bot that follows thousands of unrelated accounts.

Direction matters for input hygiene. Most triangle definitions ignore edge direction, so an edge list with both (a, b) and (b, a), or with duplicate rows, will multiply counts unless you canonicalize first. This is the most common source of wrong answers, and it produces numbers that look plausible.

From node iterator to degree ordering

The simplest method, the node iterator, looks at every vertex, enumerates every pair of its neighbours, and checks whether that pair is an edge. A pair of neighbours with a shared centre is called a wedge, or an open path of length two. The work is the number of wedges, the sum of d(v)(d(v) - 1) / 2. One vertex with a million neighbours alone produces about 5 x 10^11 wedges, which is why naive counting fails on real graphs whose degree distributions are heavy-tailed.

The fix is to orient each edge from the lower-ranked endpoint to the higher-ranked one, where rank orders vertices by degree and breaks ties by id. Then only enumerate wedges among out-neighbours. Each triangle is found exactly once, at its lowest-ranked vertex, because only that vertex has both other corners as out-neighbours. The key property is that every vertex now has at most about the square root of 2m out-neighbours, where m is the number of edges: a vertex can only point to vertices of equal or higher degree, and there cannot be more than 2m / d vertices of degree at least d. A hub with a million neighbours points to almost none of them, because almost all of them have lower degree. The total work drops to O(m^1.5), which is the bound behind most distributed triangle counting, including the MapReduce algorithms of Suri and Vassilvitskii in their 2011 paper on the curse of the last reducer.

Worked example

Take five vertices and six edges: 1-2, 1-3, 2-3, 2-4, 3-4 and 4-5. By inspection there are two triangles, {1, 2, 3} and {2, 3, 4}. Degrees are d(1) = 2, d(2) = 3, d(3) = 3, d(4) = 3 and d(5) = 1.

Rank by (degree, id): 5, 1, 2, 3, 4. Orient every edge from lower to higher rank: 1 to 2, 1 to 3, 2 to 3, 2 to 4, 3 to 4, and 5 to 4. Out-neighbour sets are 1: {2, 3}, 2: {3, 4}, 3: {4}, 4: {}, 5: {4}. Only vertices with at least two out-neighbours produce wedges. Vertex 1 produces the wedge (2, 3), and edge 2-3 exists, so that is triangle {1, 2, 3}. Vertex 2 produces (3, 4), and edge 3-4 exists, so that is triangle {2, 3, 4}. Two wedges were checked instead of the naive method's 1 + 3 + 3 + 3 + 0 = 10, and each triangle appeared once. Local counts are t(1) = 1, t(2) = 2, t(3) = 2, t(4) = 1, t(5) = 0; they sum to 6, which is 3T. The clustering coefficient of vertex 2 is 2 / 3, since two of its three neighbour pairs are connected.

The built-in operators

GraphX ships triangleCount(), which returns a graph whose vertex attribute is the local count. Since Spark 2.0 the operator canonicalizes the graph itself, removing self-loops and duplicate edges and ignoring direction; earlier versions required callers to do that, and old blog posts still say so.

import org.apache.spark.graphx._

val graph = GraphLoader
  .edgeListFile(sc, "s3://bucket/edges.txt", canonicalOrientation = true)
  .partitionBy(PartitionStrategy.RandomVertexCut)

val perVertex = graph.triangleCount().vertices      // (VertexId, Int)
val total = perVertex.map(_._2.toLong).sum() / 3    // every triangle counted at 3 vertices

The GraphX implementation builds a neighbour set for every vertex, ships the sets to the edge partitions, and intersects the two endpoint sets for each edge. That is simple and correct, but every edge incident to a hub carries the hub's full neighbour set, so memory and shuffle grow with the square of the hub's degree. Note also that the per-vertex attribute is an Int; sum it as Long.

GraphFrames exposes the same idea on DataFrames: g.triangleCount.run() returns the vertices DataFrame with a count column. It is the easiest path from PySpark, and it is fine for graphs without extreme hubs. When hubs dominate, you want control over orientation and partitioning, which means writing the algorithm directly.

A degree-ordered DataFrame implementation

The DataFrame version below follows the degree-ordered algorithm. Every step is a standard relational operator, so Spark's planner, adaptive query execution and monitoring all apply.

from pyspark.sql import functions as F

raw = spark.read.parquet("s3://bucket/edges/")          # columns: src, dst

# 1. Canonicalize: undirected, no self-loops, no duplicates.
edges = (raw.where(F.col("src") != F.col("dst"))
            .select(F.least("src", "dst").alias("u"), F.greatest("src", "dst").alias("v"))
            .distinct()
            .cache())

# 2. Degrees.
deg = (edges.select(F.col("u").alias("x"))
            .unionAll(edges.select(F.col("v").alias("x")))
            .groupBy("x").agg(F.count("*").alias("d")))

# 3. Orient each edge from lower (degree, id) to higher.
e = (edges.join(deg.select(F.col("x").alias("u"), F.col("d").alias("du")), "u")
          .join(deg.select(F.col("x").alias("v"), F.col("d").alias("dv")), "v"))
u_low = (F.col("du") < F.col("dv")) | ((F.col("du") == F.col("dv")) & (F.col("u") < F.col("v")))
oriented = e.select(F.when(u_low, F.col("u")).otherwise(F.col("v")).alias("lo"),
                    F.when(u_low, F.col("v")).otherwise(F.col("u")).alias("hi")).cache()

# 4. Wedges: pairs of out-neighbours sharing a low endpoint.
a = oriented.select(F.col("lo").alias("c"), F.col("hi").alias("x"))
b = oriented.select(F.col("lo").alias("c"), F.col("hi").alias("y"))
wedges = a.join(b, "c").where(F.col("x") < F.col("y"))

# 5. Close each wedge against the canonical edge set (u < v by id).
tri = wedges.join(edges, (F.col("x") == F.col("u")) & (F.col("y") == F.col("v"))) \
            .select("c", "x", "y")

total = tri.count()
local = (tri.select(F.explode(F.array("c", "x", "y")).alias("id"))
            .groupBy("id").agg(F.count("*").alias("triangles")))

Two details carry the correctness. The wedge pair is ordered by id with x < y so each unordered pair appears once, and the closing join uses the canonical edge set, which is also ordered by id, so the closing edge is found regardless of how it was oriented by rank. To produce zeros for vertices with no triangles, left join the vertex list to local and fill nulls with 0.

Scaling, skew and sampling

Before running step 4, compute how many wedges it will produce. The count is cheap and tells you whether the job will fit.

outdeg = oriented.groupBy("lo").agg(F.count("*").alias("k"))
wedge_rows = outdeg.select(F.sum(F.col("k") * (F.col("k") - 1) / 2)).first()[0]
max_k = outdeg.agg(F.max("k")).first()[0]
print(f"wedges={wedge_rows:,.0f}  max out-degree={max_k}")

Compare the wedge count with the number of rows your cluster comfortably shuffles, and compare the maximum out-degree with the median. With degree ordering, the maximum out-degree should be far below the maximum degree. If one vertex still dominates, its wedges all land in one task of the self-join, the classic straggler. Adaptive query execution's skew-join handling can split oversized partitions; for extreme cases, split the hot vertices' out-neighbour lists into buckets and join bucket pairs, the same salting idea described in Spark salting joins and Spark data skew.

Cache edges and oriented because both are reused, and persist tri if you need both the global and local outputs, or Spark will recompute the expensive joins for each action. Write intermediate results to storage on very large runs so a failed late stage does not repeat the early ones.

When an exact answer is not required, sample. Keep each edge independently with probability p, count triangles in the sample, and divide by p cubed, since a triangle survives only if all three edges do. This estimator, known from the DOULION work of Tsourakakis and colleagues, is unbiased and cuts the work dramatically; its variance is high for graphs with few triangles, so run it several times and report the spread.

Failure modes

Triangle jobs fail in a small number of repeatable ways.

• Inflated counts. Duplicate or reciprocal edges were not removed. Guard: canonicalize and assert that the edge count does not change after a second distinct.
• Wrong divisor. Summing local counts and forgetting to divide by 3, or dividing by 6 when the method already counts each triangle once. Guard: test on a tiny graph with a known answer, such as the worked example above or a complete graph on n vertices, which has n(n - 1)(n - 2) / 6 triangles.
• One task runs for hours. Wedge skew at a hub. Guard: degree ordering, the wedge estimate, and salting for remaining hot vertices.
• Executor out-of-memory in GraphX. Hub neighbour sets copied to many edge partitions. Guard: use the DataFrame version or remove known hubs and count them separately.
• Integer overflow. Per-vertex counts above two billion in Int columns. Guard: use long types.
• Collecting to the driver. Calling collect() on local counts for a billion vertices. Guard: write results to a table.

Operating it

Validate before scaling: sample a connected subgraph of a few hundred thousand edges, count it with networkx's nx.triangles and with your Spark job, and require exact agreement. Record the wedge count, the maximum out-degree and stage durations for each production run, and alert when the wedge count grows much faster than the edge count, which signals a new hub or a data bug such as a placeholder id that thousands of rows point at. Run triangle counts as scheduled batch jobs over snapshots; incremental maintenance is possible but rarely worth the complexity in Spark. Downstream, combine local counts with degree to compute clustering coefficients, and use them as features for ranking or fraud models, the same way PageRank scores are used.

Trade-offs

What to do next

• Canonicalize your edge list and record edge counts before and after.
• Compute degrees and the degree-ordered wedge estimate before running any triangle job.
• Validate the job against networkx on a sampled subgraph and on a complete graph with a known count.
• Start with GraphFrames for small graphs; switch to the degree-ordered DataFrame version when hubs appear.
• Watch the self-join stage for stragglers and apply salting to hot vertices if needed.
• Store local counts and clustering coefficients as features, and alert on sudden wedge growth.