Github user ilganeli commented on the issue:
https://github.com/apache/spark/pull/16685
@gatorsmile That makes a lot of sense. Here is a code snippet that relies
on the database to do the UPSERT:
```
/**
* Generate the SQL statement to perform an upsert (UPDATE OR INSERT) of
a given row into a specific table
*
* @param row The row to insert into the table
* @param schema The table schema
* @param tableName The table name in the database
* @param primaryKeys The unique constraint imposed on the database
* @return
*/
@transient
def genUpsertScript(row: Row, schema: StructType, tableName: String,
primaryKeys: Set[String]): String = {
val primaryKeyString: String = getKeyString(primaryKeys)
val schemaString = schema.map(s => s.name).reduce(_ + ", " + _)
val valString = row.toSeq.map(v => "'" + v.toString.replaceAll("'",
"''") + "'").reduce(_ + "," + _)
val withExcluded = {
schema.map(_.name)
.filterNot(primaryKeys.contains)
.map(s => s + " = EXCLUDED." + s) //EXCLUDED is a magic internal
Postgres table
.reduce(_ + ",\n" + _)
}
val upsert = {
s"INSERT INTO $tableName ($schemaString)\n VALUES ($valString)\n" +
s"ON CONFLICT ($primaryKeyString) DO UPDATE\n" +
s"SET\n" + withExcluded + ";"
}
logS("Generated SQL: " + upsert, Level.DEBUG)
upsert
}
```
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
