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

Show file tree
Hide file tree
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
Next Next commit
remove struct name from columns according to name convention in Spark… 
… 1.6 and Spark 2.x
  • Loading branch information
@estebandonato
estebandonato committed May 10, 2017
commit d93cfbfbf2510e99db3c75fae80778d2b7bd2f73
29 changes: 24 additions & 5 deletions 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 @@ -102,17 +102,36 @@ class AggregateMessages private[graphframes] (private val g: GraphFrame)
* }}}
*/
def agg(aggCol: Column, aggCols: Column*): DataFrame = {
def removeColumnNamePrefix(columnName: String) = columnName match {
case cn if cn.startsWith(s"${GraphFrame.SRC}.") => cn.substring(s"${GraphFrame.SRC}.".length)
case cn if cn.startsWith(s"${GraphFrame.DST}.") => cn.substring(s"${GraphFrame.DST}.".length)
case cn if cn.startsWith(s"${GraphFrame.EDGE}.") => cn.substring(s"${GraphFrame.EDGE}.".length)
def containsStructNameForSpark2(columnName: String, structName: String) =
columnName.startsWith(s"${structName}.")
def containsStructNameForSpark16(columnName: String, structName: String) =
columnName.startsWith(s"${structName}[")
def removeStructNameForSpark2(columnName: String, structName: String) =
columnName.substring(s"${structName}.".length)
def removeStructNameForSpark16(columnName: String, structName: String) =
columnName.substring(s"${structName}[".length, columnName.length - 1)

def removeStructName(columnName: String) = columnName match {
case cn if containsStructNameForSpark16(cn, s"${GraphFrame.SRC}") =>
removeStructNameForSpark16(cn, s"${GraphFrame.SRC}")
case cn if containsStructNameForSpark16(cn, s"${GraphFrame.DST}") =>
removeStructNameForSpark16(cn, s"${GraphFrame.DST}")
case cn if containsStructNameForSpark16(cn, s"${GraphFrame.EDGE}") =>
removeStructNameForSpark16(cn, s"${GraphFrame.EDGE}")
case cn if containsStructNameForSpark2(cn, s"${GraphFrame.SRC}") =>
removeStructNameForSpark2(cn, s"${GraphFrame.SRC}")
case cn if containsStructNameForSpark2(cn, s"${GraphFrame.DST}") =>
removeStructNameForSpark2(cn, s"${GraphFrame.DST}")
case cn if containsStructNameForSpark2(cn, s"${GraphFrame.EDGE}") =>
removeStructNameForSpark2(cn, s"${GraphFrame.EDGE}")
case cn => cn
}
def msgColumn(df: DataFrame) = df.columns.filter(_ != ID) match {
case Array(c) => df.withColumnRenamed(c, AggregateMessages.MSG_COL_NAME)
case columns => df.select(
df(ID),
struct(columns.map(c => df(s"`${c}`").as(removeColumnNamePrefix(c))) :_*).as(AggregateMessages.MSG_COL_NAME))
struct(columns.sorted.map(c => col(s"`${c}`").as(removeStructName(c))) :_*)
.as(AggregateMessages.MSG_COL_NAME))
}
require(msgToSrc.nonEmpty || msgToDst.nonEmpty, s"To run GraphFrame.aggregateMessages," +
s" messages must be sent to src, dst, or both. Set using sendToSrc(), sendToDst().")
Expand Down
4 changes: 2 additions & 2 deletions src/test/scala/org/graphframes/lib/AggregateMessagesSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -89,9 +89,9 @@ class AggregateMessagesSuite extends SparkFunSuite with GraphFrameTestSparkConte
val g = GraphFrame(vertices, edges)
val agg = g.aggregateMessages
.sendToDst(AM.src("att1"))
.sendToSrc(AM.dst("att1"))
.sendToDst(AM.src("att2"))
.sendToSrc(AM.dst("att2"))
.sendToDst(AM.src("att2"))
.sendToSrc(AM.dst("att1"))
.agg(
sum(AM.msg("att1")).as("sum_att1"),
avg(AM.msg("att2")).as("avg_att2"))
Expand Down