Skip to content

feat: aggregatemessages multiple columns #701

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
SemyonSinchenko merged 12 commits into from
Sep 23, 2025
Merged

feat: aggregatemessages multiple columns #701

Changes from 1 commit
Commits
Show all changes
12 commits
Select commit Hold shift + click to select a range
a0c302b
AggregateMessages: allow multiple calls to sendToSrc and sendToDst me…
estebandonato Apr 24, 2017
24bf147
Merge branch 'master' into aggregatemessages-multiple-columns
estebandonato May 9, 2017
d93cfbf
remove struct name from columns according to name convention in Spark…
estebandonato May 10, 2017
336804d
adapt aggregateMessages for Python to allow sending multiple columns …
estebandonato May 18, 2017
06e4369
Merge branch 'master' into aggregatemessages-multiple-columns
estebandonato May 18, 2017
e81a254
adapt test to python 3. Fix some bugs after rebasing
estebandonato May 18, 2017
401bb9f
add type validation in python api. Update documentation in python method
estebandonato May 20, 2017
60c693e
allow multiple SQL expressions when sending messages and aggregation …
estebandonato May 21, 2017
9e180de
Merge branch 'master' into aggregatemessages-multiple-columns
estebandonato Oct 7, 2017
dc6b0df
remove support for spark 1.6 + some api refactoring
estebandonato Oct 15, 2017
a17da24
Try to update 186
SemyonSinchenko Sep 22, 2025
4e0e003
compatibility with Python API
SemyonSinchenko Sep 22, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
compatibility with Python API
  • Loading branch information
@SemyonSinchenko
SemyonSinchenko committed Sep 22, 2025
commit 4e0e0035a6991da2310bf7400a988cf75fb557a5
24 changes: 24 additions & 0 deletions core/src/main/scala/org/graphframes/lib/AggregateMessages.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -77,6 +77,16 @@ class AggregateMessages private[graphframes] (private val g: GraphFrame)
this
}

// Python API compatibility
def sendToSrc(value: Column): this.type = {
msgToSrc :+= value
this
}
def sendToSrc(value: String): this.type = {
msgToSrc :+= expr(value)
this
}

/** Send message to source vertex, specifying SQL expression as a String */
def sendToSrc(value: String, values: String*): this.type =
sendToSrc(expr(value), values.map(expr): _*)
Expand All @@ -89,6 +99,16 @@ class AggregateMessages private[graphframes] (private val g: GraphFrame)
this
}

// Python API compatibility
def sendToDst(value: String): this.type = {
msgToDst :+= expr(value)
this
}
def sendToDst(value: Column): this.type = {
msgToDst :+= value
this
}

/** Send message to destination vertex, specifying SQL expression as a String */
def sendToDst(value: String, values: String*): this.type =
sendToDst(expr(value), values.map(expr): _*)
Expand Down Expand Up @@ -156,6 +176,10 @@ class AggregateMessages private[graphframes] (private val g: GraphFrame)
cachedResult
}

// Python compatibility
def agg(aggCol: Column): DataFrame = agg(aggCol, Seq.empty[Column]: _*)
def agg(aggCol: String): DataFrame = agg(expr(aggCol), Seq.empty[Column]: _*)

/**
* Run the aggregation, specifying SQL expression as a String
*
Expand Down