graphframes · SemyonSinchenko · Apr 7, 2026 · Mar 28, 2026 · Mar 28, 2026 · Apr 4, 2026
diff --git a/core/src/main/scala/org/graphframes/GraphFrame.scala b/core/src/main/scala/org/graphframes/GraphFrame.scala
@@ -236,6 +236,26 @@ class GraphFrame private (
     GraphFrame(vertices, newEdges.select(newColumns: _*))
   }
 
+  /**
+   * Reverses the direction of all edges in the graph. For every directed edge (src, dst), the
+   * resulting graph will contain an edge (dst, src) with the same attributes.
+   *
+   * @return
+   *   a new GraphFrame with all edge directions reversed.
+   *
+   * @group utils
+   */
+  def asReversed(): GraphFrame = {
+    val newEdges = edges
+      .select(col(DST).alias(SRC), col(SRC).alias(DST), nestAsCol(edges, ATTR))
+      .select(SRC, DST, ATTR)
+    val newColumns = Seq(col(SRC), col(DST)) ++ edges.columns
+      .filter(c => (c != SRC) && (c != DST))
+      .map(c => col(ATTR).getField(c).alias(c))
+      .toSeq
+    GraphFrame(vertices, newEdges.select(newColumns: _*))
+  }
+
   // ============== Basic structural methods ============
 
   /**

diff --git a/core/src/test/scala/org/graphframes/GraphFrameSuite.scala b/core/src/test/scala/org/graphframes/GraphFrameSuite.scala
@@ -682,6 +682,37 @@ class GraphFrameSuite extends SparkFunSuite with GraphFrameTestSparkContext {
     assert(edges(3).getLong(1) === 2L)
   }
 
+  test("reverse directed graph edges") {
+    val v = spark.createDataFrame(Seq((1L, "a"), (2L, "b"), (3L, "c"))).toDF("id", "name")
+    val e = spark.createDataFrame(Seq((1L, 2L), (2L, 3L))).toDF("src", "dst")
+    val g = GraphFrame(v, e)
+    val reversed = g.asReversed()
+
+    // Check edge count is the same
+    assert(reversed.edges.count() === g.edges.count())
+
+    // Verify edges are reversed
+    val edges = reversed.edges.sort("src", "dst").collect()
+    assert(edges.length === 2)
+    assert(edges(0).getLong(0) === 2L)
+    assert(edges(0).getLong(1) === 1L)
+    assert(edges(1).getLong(0) === 3L)
+    assert(edges(1).getLong(1) === 2L)
+  }
+
+  test("reverse directed graph edges with attributes") {
+    val v = spark.createDataFrame(Seq((1L, "a"), (2L, "b"))).toDF("id", "name")
+    val e = spark.createDataFrame(Seq((1L, 2L, "edge1"))).toDF("src", "dst", "attr")
+    val g = GraphFrame(v, e)
+    val reversed = g.asReversed()
+
+    val edges = reversed.edges.collect()
+    assert(edges.length === 1)
+    assert(edges(0).getLong(0) === 2L)
+    assert(edges(0).getLong(1) === 1L)
+    assert(edges(0).getString(2) === "edge1")
+  }
+
   test("toGraphX should throw IllegalArgumentException for null IDs") {
     val schema = StructType(
       Seq(

diff --git a/python/graphframes/graphframe.py b/python/graphframes/graphframe.py
@@ -1012,6 +1012,31 @@ def as_undirected(self) -> "GraphFrame":
 
         return GraphFrame(self.vertices, new_edges)
 
+    def as_reversed(self) -> "GraphFrame":
+        """
+        Reverses the direction of all edges in the graph. For every directed edge (src, dst),
+        the resulting graph will contain an edge (dst, src) with the same attributes.
+
+        :return: A new GraphFrame with all edge directions reversed.
+        """
+        edge_attr_columns = [c for c in self.edges.columns if c not in [SRC, DST]]
+
+        if edge_attr_columns:
+            reversed_edges = self.edges.select(
+                F.col(DST).alias(SRC),
+                F.col(SRC).alias(DST),
+                F.struct(*edge_attr_columns).alias(EDGE),
+            )
+            reversed_edges = reversed_edges.select(SRC, DST, EDGE)
+        else:
+            reversed_edges = self.edges.select(F.col(DST).alias(SRC), F.col(SRC).alias(DST))
+
+        edge_columns = [F.col(EDGE).getField(c).alias(c) for c in edge_attr_columns]
+        selected_columns = [F.col(SRC), F.col(DST)] + edge_columns
+        reversed_edges = reversed_edges.select(*selected_columns)
+
+        return GraphFrame(self.vertices, reversed_edges)
+
     def aggregate_neighbors(
         self,
         starting_vertices: Column | str,

diff --git a/python/tests/test_graphframes.py b/python/tests/test_graphframes.py
@@ -145,6 +145,37 @@ def test_as_undirected(spark: SparkSession) -> None:
     assert any(row[0] == 2 and row[1] == 1 and row[2] == "edge1" for row in edges2)
 
 
+def test_as_reversed(spark: SparkSession) -> None:
+    # Test without edge attributes
+    v = spark.createDataFrame([(1, "a"), (2, "b"), (3, "c")]).toDF("id", "name")
+    e = spark.createDataFrame([(1, 2), (2, 3)]).toDF("src", "dst")
+    g = GraphFrame(v, e)
+    reversed_g = g.as_reversed()
+
+    # Check edge count is the same
+    assert reversed_g.edges.count() == g.edges.count()
+
+    # Verify edges are reversed
+    edges = reversed_g.edges.sort("src", "dst").collect()
+    assert len(edges) == 2
+    assert edges[0][0] == 2
+    assert edges[0][1] == 1
+    assert edges[1][0] == 3
+    assert edges[1][1] == 2
+
+    # Test with edge attributes
+    v2 = spark.createDataFrame([(1, "a"), (2, "b")]).toDF("id", "name")
+    e2 = spark.createDataFrame([(1, 2, "edge1")]).toDF("src", "dst", "attr")
+    g2 = GraphFrame(v2, e2)
+    reversed2 = g2.as_reversed()
+
+    edges2 = reversed2.edges.collect()
+    assert len(edges2) == 1
+    assert edges2[0][0] == 2
+    assert edges2[0][1] == 1
+    assert edges2[0][2] == "edge1"
+
+
 def test_cache(local_g: GraphFrame) -> None:
     _ = local_g.cache()
     _ = local_g.unpersist()