Is there a way to define the relationship type using a dataframe column in spark?

Is there a way to define the relationship type using a dataframe column in spark?

import pandas as pd

_list = []
_dict = {}

_dict['ENV'] = "DEV"
_dict['PRIVILEGE'] = "DML"
_dict['ROLE'] = "ROLE1"
_dict['DATABASE'] = "Database1"
_list.append(_dict)

_dict['ENV'] = "DEV"
_dict['PRIVILEGE'] = "DDL"
_dict['ROLE'] = "ROLE2"
_dict['DATABASE'] = "Database1"
_list.append(_dict)

df = pd.DataFrame(_list)
df = spark.createDataFrame(df)

df.write.format("org.neo4j.spark.DataSource") \
  .mode("Overwrite")  \
  .option("relationship", df["PRIVILEGE"]) \
  .option("relationship.save.strategy", "keys") \
  .option("relationship.target.labels", "DATABASE") \
  .option("relationship.target.node.keys", "DATABASE, ENV") \
  .option("relationship.target.save.mode", "overwrite") \
  .option("relationship.source.labels", "ROLE") \
  .option("relationship.source.save.mode", "overwrite") \
  .option("relationship.source.node.keys", "ROLE, ENV") \
  .save()

It is writing the type as Column<'PRIVILEGE'>, when I want the value.

df["PRIVILEGE"] returns a org.apache.spark.sql.Column, so what you are using as the relationship type is just the value of Column.toString() (i.e., Column<'PRIVILEGE'>).

The "relationship" option expects its argument (the relationship type) to be a fixed string. It does not support getting the relationship type name dynamically from a DataFrame column.

Therefore, you should create a separate DataFrame for each relationship type and have a separate write for each DataFrame.

And, by the way, since you reuse the same _dict instance for every _list element, all _list elements will be identical. You need to create a new dict for every element.

Please do raise an issue so the team can consider how it can be implemented

The approaches i've seen be used are to either:

• iterate one relationship type at a time (as mentioned by @cybersam and @jose_bacoy)

e.g. get unique values of the PRIVILEGE column = relationship types

relTypeColumn="PRIVILEGE"
uniquerelTypes=df.dropDuplicates([relTypeColumn])
relTypes = uniquerelTypes.select(relTypeColumn).collect()

Then iterate over those relationship types, filter the DataFrame for those rows that match and write to Neo4j

for row in relTypes:
    relType=row[relTypeColumn]
    relsubset=df.filter(df[relTypeColumn] == relType)
    relsubset.write.format("org.neo4j.spark.DataSource") \
    .mode("Overwrite")  \
    .option("relationship", relType) \
    .option("relationship.save.strategy", "keys") \
    .option("relationship.target.labels", "DATABASE") \
    .option("relationship.target.node.keys", "DATABASE, ENV") \
    .option("relationship.target.save.mode", "overwrite") \
    .option("relationship.source.labels", "ROLE") \
    .option("relationship.source.save.mode", "overwrite") \
    .option("relationship.source.node.keys", "ROLE, ENV") \
    .option("url", neo4j_url) \
    .option("authentication.type", "basic") \
    .option("authentication.basic.username", neo4j_user) \
    .option("authentication.basic.password", neo4j_password) \
    .save()

• use apoc.create.relationship or apoc.merge.relationship within a query

e.g. write nodes

df.write.format("org.neo4j.spark.DataSource") \
.mode("Overwrite")  \
.option("labels", ":DATABASE") \
.option("node.keys", "DATABASE,ENV") \
.option("schema.optimization.type", "NODE_CONSTRAINTS") \
.option("url", neo4j_url) \
.option("authentication.type", "basic") \
.option("authentication.basic.username", neo4j_user) \
.option("authentication.basic.password", neo4j_password) \
.save()

Then write the relationships using apoc to allow passing in dynamic values from the DataFrame

df.write.format("org.neo4j.spark.DataSource") \
.mode("Overwrite")  \
.option("query", """
MATCH (database:DATABASE {DATABASE:event.DATABASE, ENV:event.ENV}) 
MATCH (role:ROLE {ROLE:event.ROLE, ENV:event.ENV})
WITH database, role, event
CALL apoc.create.relationship(database, event.PRIVILEGE, {}, role) YIELD rel
return count(role)""") \
.option("url", neo4j_url) \
.option("authentication.type", "basic") \
.option("authentication.basic.username", neo4j_user) \
.option("authentication.basic.password", neo4j_password) \
.save()