MERGE/SET nodes with multiple properties using Neo4j Spark Connector about neo4j-spark-connector

[sjishan]

Solution

val source: (Seq[String], Seq[String]) = (Seq("uid"), Seq("unix_timestamp_s") )
val target: (Seq[String], Seq[String]) = (Seq("vid"), Seq("unix_timestamp_s"))

val mergeStatement = s"""
      UNWIND {rows} as row
      MERGE (source:ID {id : row.srcid.`${source._1.head}`})
      SET source.timestamp = row.srctmp.`${source._2.head}`, source.type= 'abc'
      MERGE (target:ID {id : row.trgid.`${target._1.head}`})
      SET target.timestamp = row.trgtmp.`${target._2.head}`, target.type= 'xyz'
      MERGE (source)-[:HasConnection]-(target)
      """

df.repartition($"component").foreachPartition( rows => {
      val params: AnyRef = rows.map(r =>
        Map(
           "srcid" -> source._1.map( c => (c, r.getAs[AnyRef](c))).toMap.asJava,
           "srctmp" -> source._2.map( c => (c, r.getAs[AnyRef](c))).toMap.asJava,
           "trgid" -> target._1.map( c => (c, r.getAs[AnyRef](c))).toMap.asJava,
           "trgtmp" -> target._2.map( c => (c, r.getAs[AnyRef](c))).toMap.asJava
        )
          .asJava).asJava
          execute(config, mergeStatement, Map("rows" -> params).asJava)
    })