Avez-vous déjà vu... un DataFrame Spark ayant deux colonnes ayant le même nom ? Et avez-vous essayé de lancer une requête SQL dessus ?
Il est possible que ça soit le cas si vous utilisez Spark, même de manière sporadique. Lorsque vous faites une jointure entre deux dataframes, vous allez rencontrer rapidement ce cas de figure :
case class Person(id: String, name: String, age: Int)
case class Status(id: String, status: String)
val statusDF: DataFrame = ss.createDataFrame(Seq(Person(id = "id1", name = "toto", age = 13)))
val personDF: DataFrame = ss.createDataFrame(Seq(Status(id = "id1", status = "active")))
val dataframe: DataFrame =
personDF.join(statusDF, personDF("id") === statusDF("id"))
dataframe.select("id", "name", "status").show()Lors du select, Spark va vous rappeler gentiment à l'ordre avec l'exception suivante :
org.apache.spark.sql.AnalysisException: Reference 'id' is ambiguous, could be: id, id.;Mais, vous pouvez aussi avoir la maladresse d'enregistrer un tel DataFrame sur disque (en utilisant le format Parquet, par exemple). Sans le savoir, vous allez vous retrouver avec une colonne illisible. Vous allez du coup être accueillis par une autre exception lors de la lecture :
java.lang.RuntimeException: [id] BINARY was added twiceAlors que faire dans ce cas-là ? Sommes-nous condamnés à maudire la personne qui a osé mettre deux noms de colonne identiques ? 🤬
Vous pouvez vous en sortir avec trois solutions :
- Il existe une solution très facile à ce problème, en modifiant le nom de la colonne responsable (ici
id) en amont de la jointure :
val statusDFRenamed: DataFrame = statusDF.withColumnRenamed("id", "idStatus")
//ou bien
val personDFRenamed: DataFrame = personDF.withColumnRenamed("id", "idPerson")- Si vous souhaitez travailler directement sur la dataframe en sortie de jointure, vous pouvez utiliser une fonction que nous avons mis dans Plumbus, qui réalise un
coalesceentre des colonnes de même nom. Pour réussir à définir cette transformation, on passe par la structure de la requête (Catalyst), afin d'adresser individuellement les colonnes qui ont le même nom (NamedExpression).
def coalesceColWithSameName(df: DataFrame): DataFrame = {
//pour être sûr que c'est un Project
val frame: DataFrame = df.select("*")
//récupération de toutes les expressions : id, name, age, id, status
val cols: Seq[NamedExpression] = frame.queryExecution.analyzed.asInstanceOf[Project].projectList
//colonnes en sortie, dans le même ordre, sans les doublons
val outputCols: Array[String] = frame.columns.distinct
import org.apache.spark.sql.functions.coalesce
val colMap: Map[String, Column] = {
val nameExpressionsByName: Map[String, Seq[NamedExpression]] = cols.groupBy(_.name)
nameExpressionsByName.map({
//seulement une expression pour ce nom
case (name, Seq(expr)) => (name, new Column(expr).as(name))
//plusieurs expressions pour ce nom => fusion avec coalesce
case (name, nameExpressions) => (name, coalesce(nameExpressions.map(x => new Column(x)): _*).as(name))
})
}
frame.select(outputCols.map(colMap): _*)
}- Vous pouvez utiliser une autre solution que nous avons aussi ajouté à Plumbus, avec un comportement différent. Ici, les colonnes en doublon vont être renommées séparément avec un suffixe généré automatiquement (
_1, _2, _3, ..., _n). Même technique, nous utilisons la structure de la requête pour adresser individuellement les colonnes.
def renameColumnsWithSameName(df: DataFrame): DataFrame = {
val frame: DataFrame = df.select("*")
val namedExpressions: Seq[NamedExpression] = frame.queryExecution.analyzed.asInstanceOf[Project].projectList
val duplicateNames: Set[String] = namedExpressions.map(_.name).groupBy(identity).filter(_._2.size > 1).keys.toSet
if (duplicateNames.isEmpty) {
df
} else {
type NameCount = Map[String, Int]
type Res = (NameCount, Seq[Column])
val zero: Res = (Map.empty, Nil)
val res = namedExpressions.foldLeft(zero)({
case ((counts, cols), exp) =>
if (duplicateNames(exp.name)) {
val i = counts.getOrElse(exp.name, 0)
val col = new Column(exp).as(exp.name + "_" + i)
(counts + (exp.name -> (i + 1)), cols :+ col)
} else {
(counts, cols :+ new Column(exp))
}
})
df.select(res._2: _*)
}
}Voici un exemple :
import ss.implicits._
val person: Person = Person(id = "1", name = "toto", age = 13)
val status: Status = Status(id = "1", status = "active")
val df1: Dataset[Person] = ss.createDataset(Seq(person))
val df2: Dataset[Status] = ss.createDataset(Seq(status))
val frame: DataFrame = df1.join(df2, df1("id") === df2("id"))
println("initial DataFrame")
frame.show()
//clean auto1
println("coalesceColWithSameName")
coalesceColWithSameName(frame).show()
//clean auto2
println("renameColWithSameName")
renameColumnsWithSameName(frame).show()Nous obtenons en sortie :
initial DataFrame
+---+----+---+---+------+
| id|name|age| id|status|
+---+----+---+---+------+
| 1|toto| 13| 1|active|
+---+----+---+---+------+
coalesceColWithSameName
+---+----+---+------+
| id|name|age|status|
+---+----+---+------+
| 1|toto| 13|active|
+---+----+---+------+
renameColWithSameName
+----+----+---+----+------+
|id_0|name|age|id_1|status|
+----+----+---+----+------+
| 1|toto| 13| 1|active|
+----+----+---+----+------+Conclusion
Nous avons vu comment gérer ces dataframes assez embêtant et nous avons vu trois manières de répondre à ce problème, chacune ayant un niveau de difficulté variable. Évidemment le but de cet article était d'en apprendre plus sur Catalyst et sur les différentes opérations possibles faisable avec celui-ci. Il peut être très bénéfique d'utiliser Catalyst dans le cadre d'opérations manipulant le schéma d'un dataframe en Spark.
Nous avons vu rapidement comment résoudre le problème des doublons de colonnes de manière générique, en allant chercher un peu d'info dans les représentations internes de Spark SQL (ie. avec Catalyst). Ces APIs, assez méconnues, peuvent être très intéressantes si nous devons aller plus loin que la manipulation générique de schéma (avec df.schema).
Notes
- lorsque l'on manipule les dataframes :org.apache.spark.sql.AnalysisException: Reference 'id' is ambiguous, could be: id#3, id#7.;
- si on sauvegarde en parquet et que l'on relit ça après : Caused by: java.lang.RuntimeException: [id] BINARY was added twice
Solutions :
- à la main
- avec
def coalesceColwithSameName(df: DataFrame):DataFrame - avec
def renameColumnsWithSameName(df:DataFrame):DataFrame
Code
import org.apache.spark.sql.catalyst.expressions.NamedExpression
import org.apache.spark.sql.catalyst.plans.logical.Project
import org.apache.spark.sql.{Column, DataFrame, SparkSession}
import org.scalatest.FunSuite
case class Person(id:String,name:String,age:Int)
class TestDuplicateCols extends FunSuite with TestUtils {
val ss: SparkSession = TestSparkSession.create()
def coalesceColwithSameName(df: DataFrame):DataFrame = {
//pour être sûr que c'est un Projet
val frame = df.select("*")
//récupération de toutes les expressions : id#3, name#4, age#5,id#7, name#8, age#9
val cols: Seq[NamedExpression] = frame.queryExecution.analyzed.asInstanceOf[Project].projectList
//colonnes en sortie
val outpuCols = frame.columns.distinct
import org.apache.spark.sql.functions.coalesce
//
val colMap: Map[String, Column] = {
val nameExpressionsByName: Map[String, Seq[NamedExpression]] = cols.groupBy(_.name)
nameExpressionsByName.map({
//seulement une expression pour ce nom
case (name, Seq(expr)) => (name, new Column(expr).as(name))
//plusieurs expressions pour ce nom => fusion avec coalesce
case (name, nameExpressions) => (name, coalesce(nameExpressions.map(x => new Column(x)): _*).as(name))
})
}
frame.select(outpuCols.map(colMap):_*)
}
def renameColumnsWithSameName(df:DataFrame):DataFrame = {
val frame = df.select("*")
val namedExpressions = frame.queryExecution.analyzed.asInstanceOf[Project].projectList
val duplicateNames: Set[String] = namedExpressions.map(_.name).groupBy(identity).filter(_._2.size > 1).keys.toSet
if(duplicateNames.isEmpty) df else {
type NameCount = Map[String,Int]
type Res = (NameCount,Seq[Column])
val zero:Res = (Map.empty,Nil)
val res = namedExpressions.foldLeft(zero)({
case ((counts,cols),exp) =>
if(duplicateNames(exp.name)) {
val i = counts.getOrElse(exp.name,0)
val col = new Column(exp).as(exp.name + "_" + i)
(counts + (exp.name -> (i + 1)),cols :+ col)
} else {
(counts, cols :+ new Column(exp))
}
})
df.select(res._2:_*)
}
}
test("toto") {
import ss.implicits._
val person = Person(id = "1", name = "toto", age = 13)
val df1 = ss.createDataset(Seq(person))
val df2 = df1
val frame = df1.join(df2, df1("id") === df2("id"))
//frame.apply("xyz")
frame.show()
val plan = frame.select("*").queryExecution.analyzed
//bug
//frame.select("id","age","name")
//work
frame.select(df1("id"),df1("age"),df1("name")).show()
//clean auto1
coalesceColwithSameName(frame).show()
//clean auto2
renameColumnsWithSameName(frame).show()
}
}Output
+---+----+---+---+----+---+
| id|name|age| id|name|age|
+---+----+---+---+----+---+
| 1|toto| 13| 1|toto| 13|
+---+----+---+---+----+---+
+---+---+----+
| id|age|name|
+---+---+----+
| 1| 13|toto|
+---+---+----+
+---+----+---+
| id|name|age|
+---+----+---+
| 1|toto| 13|
+---+----+---+
+----+------+-----+----+------+-----+
|id_0|name_0|age_0|id_1|name_1|age_1|
+----+------+-----+----+------+-----+
| 1| toto| 13| 1| toto| 13|
+----+------+-----+----+------+-----+Code formaté
import org.apache.spark.sql.catalyst.expressions.NamedExpression
import org.apache.spark.sql.catalyst.plans.logical.Project
import org.apache.spark.sql.{Column, DataFrame, SparkSession}
import org.scalatest.FunSuite
case class Person(id: String, name: String, age: Int)
class TestDuplicateCols extends FunSuite {
val ss: SparkSession = SparkSession.builder().master("local[*]").getOrCreate()
def coalesceColwithSameName(df: DataFrame): DataFrame = {
//pour être sûr que c'est un Project
val frame = df.select("*")
//récupération de toutes les expressions : id#3, name#4, age#5,id#7, name#8, age#9
val cols: Seq[NamedExpression] = frame.queryExecution.analyzed.asInstanceOf[Project].projectList
//colonnes en sortie
val outpuCols = frame.columns.distinct
import org.apache.spark.sql.functions.coalesce
val colMap: Map[String, Column] = {
val nameExpressionsByName: Map[String, Seq[NamedExpression]] = cols.groupBy(_.name)
nameExpressionsByName.map({
//seulement une expression pour ce nom
case (name, Seq(expr)) => (name, new Column(expr).as(name))
//plusieurs expressions pour ce nom => fusion avec coalesce
case (name, nameExpressions) => (name, coalesce(nameExpressions.map(x => new Column(x)): _*).as(name))
})
}
frame.select(outpuCols.map(colMap): _*)
}
def renameColumnsWithSameName(df: DataFrame): DataFrame = {
val frame = df.select("*")
val namedExpressions = frame.queryExecution.analyzed.asInstanceOf[Project].projectList
val duplicateNames: Set[String] = namedExpressions.map(_.name).groupBy(identity).filter(_._2.size > 1).keys.toSet
if (duplicateNames.isEmpty) df else {
type NameCount = Map[String, Int]
type Res = (NameCount, Seq[Column])
val zero: Res = (Map.empty, Nil)
val res = namedExpressions.foldLeft(zero)({
case ((counts, cols), exp) =>
if (duplicateNames(exp.name)) {
val i = counts.getOrElse(exp.name, 0)
val col = new Column(exp).as(exp.name + "_" + i)
(counts + (exp.name -> (i + 1)), cols :+ col)
} else {
(counts, cols :+ new Column(exp))
}
})
df.select(res._2: _*)
}
}
test("toto") {
import ss.implicits._
val person = Person(id = "1", name = "toto", age = 13)
val df1 = ss.createDataset(Seq(person))
val df2 = df1
val frame = df1.join(df2, df1("id") === df2("id"))
frame.show()
val plan = frame.select("*").queryExecution.analyzed
//bug
//frame.select("id","age","name")
//work
frame.select(df1("id"), df1("age"), df1("name")).show()
//clean auto1
coalesceColwithSameName(frame).show()
//clean auto2
renameColumnsWithSameName(frame).show()
}
}