Struct pregel_rs::pregel::Pregel

pub struct Pregel { /* private fields */ }

The Pregel struct represents a Pregel computation with various parameters and expressions.

Properties:

• graph: The graph property is a GraphFrame struct that represents the graph data structure used in the Pregel algorithm. It contains information about the vertices and edges of the graph.

• max_iterations: The maximum number of iterations that the Pregel algorithm will run for.

• vertex_column: vertex_column is a property of the PregelBuilder struct that represents the name of the column in the graph’s vertex DataFrame that contains the vertex IDs. This column is used to identify and locate a column where we apply some of the provided operations during the Pregel computation.

• initial_message: initial_message is an expression that defines the initial message that each vertex in the graph will receive before the computation starts. This message can be used to initialize the state of each vertex or to provide some initial information to the computation.

• send_messages: send_messages is a tuple containing two expressions. The first expression represents the message sending function that determines whether the message will go from Src to Dst or vice-versa. The second expression represents the message sending function that determines which messages to send from a vertex to its neighbors.

• aggregate_messages: aggregate_messages is an expression that defines how messages sent to a vertex should be aggregated. In Pregel, messages are sent from one vertex to another and can be aggregated before being processed by the receiving vertex. The aggregate_messages expression specifies how these messages should be combined.

• v_prog: v_prog is an expression that defines the vertex program for the Pregel algorithm. It specifies the computation that each vertex performs during each iteration of the algorithm. The vertex program can take as input the current state of the vertex, the messages received from its neighbors or and any other relevant information.

Implementations

impl Pregel

pub fn src(column_name: ColumnIdentifier) -> Expr

This function returns an expression for a column identifier representing the source vertex in a Pregel graph.

Arguments:

• column_name: column_name is a parameter of type ColumnIdentifier. It is used to identify the name of a column in a table or data source. The src function takes this parameter and returns an expression that represents the value of the column with the given name.

Returns:

The function src returns an Expr which represents a reference to the source vertex ID column in a Pregel graph computation. The Expr is created using the col function and the alias method of the Pregel struct to generate the appropriate column name.

Examples found in repository

examples/maximum_value.rs (line 30)

fn main() -> Result<(), Box<dyn Error>> {
    let edges = df![
        Src.as_ref() => [0, 1, 1, 2, 2, 3],
        Dst.as_ref() => [1, 0, 3, 1, 3, 2],
    ]?;

    let vertices = df![
        Id.as_ref() => [0, 1, 2, 3],
        Custom("value").as_ref() => [3, 6, 2, 1],
    ]?;

    let pregel = PregelBuilder::new(GraphFrame::new(vertices, edges)?)
        .max_iterations(4)
        .with_vertex_column(Custom("max_value"))
        .initial_message(col(Custom("value").as_ref()))
        .send_messages(MessageReceiver::Dst, Pregel::src(Custom("max_value")))
        .aggregate_messages(Pregel::msg(None).max())
        .v_prog(max_exprs([
            col(Custom("max_value").as_ref()),
            Pregel::msg(None),
        ]))
        .build();

    Ok(println!("{}", pregel.run()?))
}

examples/pagerank.rs (line 31)

fn main() -> Result<(), Box<dyn Error>> {
    let edges = df![
        Src.as_ref() => [0, 0, 1, 2, 3, 4, 4, 4],
        Dst.as_ref() => [1, 2, 2, 3, 3, 1, 2, 3],
    ]?;

    let vertices = GraphFrame::from_edges(edges.clone())?.out_degrees()?;

    let damping_factor = 0.85;
    let num_vertices: f64 = vertices.column(Id.as_ref())?.len() as f64;

    let pregel = PregelBuilder::new(GraphFrame::new(vertices, edges)?)
        .max_iterations(4)
        .with_vertex_column(Custom("rank"))
        .initial_message(lit(1.0 / num_vertices))
        .send_messages(
            MessageReceiver::Dst,
            Pregel::src(Custom("rank")) / Pregel::src(Custom("out_degree")),
        )
        .aggregate_messages(Pregel::msg(None).sum())
        .v_prog(
            Pregel::msg(None) * lit(damping_factor) + lit((1.0 - damping_factor) / num_vertices),
        )
        .build();

    Ok(println!("{}", pregel.run()?))
}

pub fn dst(column_name: ColumnIdentifier) -> Expr

This function returns an expression for a column identifier representing the destination vertex in a Pregel graph.

Arguments:

• column_name: column_name is a parameter of type ColumnIdentifier which represents the name of a column in a table. It is used as an argument to the dst function to create an expression that refers to the column with the given name in the context of a Pregel computation.

Returns:

The function dst returns an expression that represents the value of the column with the given column_name in the context of a Pregel graph computation. The expression is created using the col function and the alias method of the Pregel struct to ensure that the column name is properly qualified.

pub fn edge(column_name: ColumnIdentifier) -> Expr

This function returns an expression for a column name in a graph edge table.

Arguments:

• column_name: column_name is a parameter of type ColumnIdentifier which represents the name of a column in a graph edge table. The edge function returns an expression that refers to this column using the col function and the alias function from the Pregel struct.

Returns:

The function edge returns an Expr which represents a reference to a column in a graph edge table. The column name is passed as an argument to the function and is used to construct the full column identifier using the Pregel::alias method.

pub fn msg(column_name: Option<ColumnIdentifier>) -> Expr

This function returns an expression for a column name, either using a default value or a specified value.

Arguments:

• column_name: An optional parameter of type ColumnIdentifier. It represents the name of a column in a table. If it is None, the function returns a reference to the Pregel column. If it is Some(column_name), the function returns a reference to a column with the name

Returns:

an Expr which is either a reference to the ColumnIdentifier::Pregel column if column_name is None, or a reference to a column with an alias created by Pregel::alias if column_name is Some.

Examples found in repository

examples/maximum_value.rs (line 31)

fn main() -> Result<(), Box<dyn Error>> {
    let edges = df![
        Src.as_ref() => [0, 1, 1, 2, 2, 3],
        Dst.as_ref() => [1, 0, 3, 1, 3, 2],
    ]?;

    let vertices = df![
        Id.as_ref() => [0, 1, 2, 3],
        Custom("value").as_ref() => [3, 6, 2, 1],
    ]?;

    let pregel = PregelBuilder::new(GraphFrame::new(vertices, edges)?)
        .max_iterations(4)
        .with_vertex_column(Custom("max_value"))
        .initial_message(col(Custom("value").as_ref()))
        .send_messages(MessageReceiver::Dst, Pregel::src(Custom("max_value")))
        .aggregate_messages(Pregel::msg(None).max())
        .v_prog(max_exprs([
            col(Custom("max_value").as_ref()),
            Pregel::msg(None),
        ]))
        .build();

    Ok(println!("{}", pregel.run()?))
}

examples/pagerank.rs (line 33)

fn main() -> Result<(), Box<dyn Error>> {
    let edges = df![
        Src.as_ref() => [0, 0, 1, 2, 3, 4, 4, 4],
        Dst.as_ref() => [1, 2, 2, 3, 3, 1, 2, 3],
    ]?;

    let vertices = GraphFrame::from_edges(edges.clone())?.out_degrees()?;

    let damping_factor = 0.85;
    let num_vertices: f64 = vertices.column(Id.as_ref())?.len() as f64;

    let pregel = PregelBuilder::new(GraphFrame::new(vertices, edges)?)
        .max_iterations(4)
        .with_vertex_column(Custom("rank"))
        .initial_message(lit(1.0 / num_vertices))
        .send_messages(
            MessageReceiver::Dst,
            Pregel::src(Custom("rank")) / Pregel::src(Custom("out_degree")),
        )
        .aggregate_messages(Pregel::msg(None).sum())
        .v_prog(
            Pregel::msg(None) * lit(damping_factor) + lit((1.0 - damping_factor) / num_vertices),
        )
        .build();

    Ok(println!("{}", pregel.run()?))
}

pub fn run(self) -> PolarsResult<DataFrame>

Represents the Pregel model for large-scale graph processing, introduced by Google in a paper titled “Pregel: A System for Large-Scale Graph Processing” in 2010.

The Pregel model is a distributed computing model for processing graph data in a distributed and parallel manner. It is designed for efficiently processing large-scale graphs with billions or trillions of vertices and edges.

Components

• Vertices: Represent the entities in the graph and store the local state of each entity. Vertices can perform computations and communicate with their neighboring vertices.

• Edges: Represent the relationships between vertices and are used for communication between vertices during computation.

• Computation: Each vertex performs a user-defined computation during each super-step, based on its local state and the messages received from its neighboring vertices.

• Messages: Vertices can send messages to their neighboring vertices during each super-step, which are then delivered in the next super-step. Messages are used for communication and coordination between vertices.

• Aggregators: functions that can be used to collect and aggregate information from vertices during computation. Aggregators allow for global coordination and information gathering across the entire graph.

Usage

The Pregel model follows a sequence of super-step, where each super-step consists of computation, message exchange, and aggregation. Vertices perform their computations in parallel, and messages are exchanged between vertices to coordinate their activities. The computation continues in a series of super-steps until a termination condition is met.

This Rust function provides an implementation of the Pregel model for graph processing, allowing users to run vertex-centric algorithms that operate on the local state of each vertex and communicate through messages.

Notes

• This is a simplified example of the Pregel model and may require further customization based on the specific graph processing requirements.

Returns:

a PolarsResult<DataFrame>, which is a result type that can either contain the resulting DataFrame or an error of type PolarsError.

Examples found in repository

examples/maximum_value.rs (line 38)

fn main() -> Result<(), Box<dyn Error>> {
    let edges = df![
        Src.as_ref() => [0, 1, 1, 2, 2, 3],
        Dst.as_ref() => [1, 0, 3, 1, 3, 2],
    ]?;

    let vertices = df![
        Id.as_ref() => [0, 1, 2, 3],
        Custom("value").as_ref() => [3, 6, 2, 1],
    ]?;

    let pregel = PregelBuilder::new(GraphFrame::new(vertices, edges)?)
        .max_iterations(4)
        .with_vertex_column(Custom("max_value"))
        .initial_message(col(Custom("value").as_ref()))
        .send_messages(MessageReceiver::Dst, Pregel::src(Custom("max_value")))
        .aggregate_messages(Pregel::msg(None).max())
        .v_prog(max_exprs([
            col(Custom("max_value").as_ref()),
            Pregel::msg(None),
        ]))
        .build();

    Ok(println!("{}", pregel.run()?))
}

examples/pagerank.rs (line 39)

fn main() -> Result<(), Box<dyn Error>> {
    let edges = df![
        Src.as_ref() => [0, 0, 1, 2, 3, 4, 4, 4],
        Dst.as_ref() => [1, 2, 2, 3, 3, 1, 2, 3],
    ]?;

    let vertices = GraphFrame::from_edges(edges.clone())?.out_degrees()?;

    let damping_factor = 0.85;
    let num_vertices: f64 = vertices.column(Id.as_ref())?.len() as f64;

    let pregel = PregelBuilder::new(GraphFrame::new(vertices, edges)?)
        .max_iterations(4)
        .with_vertex_column(Custom("rank"))
        .initial_message(lit(1.0 / num_vertices))
        .send_messages(
            MessageReceiver::Dst,
            Pregel::src(Custom("rank")) / Pregel::src(Custom("out_degree")),
        )
        .aggregate_messages(Pregel::msg(None).sum())
        .v_prog(
            Pregel::msg(None) * lit(damping_factor) + lit((1.0 - damping_factor) / num_vertices),
        )
        .build();

    Ok(println!("{}", pregel.run()?))
}