| 16 | @check_arg_types |
| 17 | @trace_user_frame |
| 18 | def pagerank(edges: pw.Table[Edge], steps: int = 5) -> pw.Table[Result]: |
| 19 | in_vertices: pw.Table = edges.groupby(id=edges.v).reduce(degree=0) |
| 20 | out_vertices: pw.Table = edges.groupby(id=edges.u).reduce( |
| 21 | degree=pw.reducers.count() |
| 22 | ) |
| 23 | degrees: pw.Table = pw.Table.update_rows(in_vertices, out_vertices) |
| 24 | base: pw.Table = out_vertices.difference(in_vertices).select(rank=1_000) |
| 25 | |
| 26 | ranks: pw.Table = degrees.select(rank=6_000) |
| 27 | |
| 28 | for step in range(steps): |
| 29 | outflow = degrees.select( |
| 30 | flow=pw.if_else( |
| 31 | degrees.degree == 0, 0, (ranks.rank * 5) // (degrees.degree * 6) |
| 32 | ), |
| 33 | ) |
| 34 | |
| 35 | inflows = edges.groupby(id=edges.v).reduce( |
| 36 | rank=pw.reducers.sum(outflow.ix(edges.u).flow) + 1_000 |
| 37 | ) |
| 38 | |
| 39 | ranks = pw.Table.concat(base, inflows).with_universe_of(degrees) |
| 40 | |
| 41 | return ranks |