Aller au contenu principal
Dernière mise à jour le

Fonctions définies par l'utilisateur dans Databricks Connect pour Scala

remarque

Cet article couvre Databricks Connect pour Databricks Runtime 14.1 et les versions ultérieures.

Databricks Connect pour Scala prend en charge l'exécution de fonctions définies par l'utilisateur (UDF) sur les clusters Databricks depuis votre environnement de développement local.

Cette page décrit comment exécuter des fonctions définies par l'utilisateur avec Databricks Connect pour Scala.

Pour la version Python de cet article, consultez Fonctions définies par l'utilisateur dans Databricks Connect pour Python.

upload compiled class et des fichiers JAR​

Pour que les UDF fonctionnent, les classes compilées et les JARs doivent être upload au cluster à l'aide de l'API addCompiledArtifacts().

remarque

La version Scala utilisée par le client doit correspondre à la version Scala sur le cluster Databricks. Pour vérifier la version Scala du cluster, consultez la section « Environnement système » pour la version de Databricks Runtime du cluster dans Versions et compatibilité des notes de version de Databricks Runtime.

Le programme Scala suivant configure une simple UDF qui élève les valeurs d'une colonne au carré.

Scala
import com.databricks.connect.DatabricksSession
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions.{col, udf}

object Main {
  def main(args: Array[String]): Unit = {
    val spark = getSession()

    val squared = udf((x: Long) => x * x)

    spark.range(3)
      .withColumn("squared", squared(col("id")))
      .select("squared")
      .show()

    }
  }

  def getSession(): SparkSession = {
    if (sys.env.contains("DATABRICKS_RUNTIME_VERSION")) {
      // On a Databricks cluster — reuse the active session
      SparkSession.active
    } else {
      // Locally with Databricks Connect — upload local JARs and classes
      DatabricksSession
        .builder()
        .addCompiledArtifacts(
          Main.getClass.getProtectionDomain.getCodeSource.getLocation.toURI
        )
        .getOrCreate()
    }
  }
}

Main.getClass.getProtectionDomain.getCodeSource.getLocation.toURI pointe vers le même emplacement que la sortie compilée du projet (par exemple, target/classes ou le JAR compilé). Toutes les classes compilées sont téléchargées vers Databricks, pas seulement Main.

target/scala-2.13/classes/
├── com/
│   ├── examples/
│   │   ├── Main.class
│   │   └── MyUdfs.class
│   └── utils/
│       └── Helper.class

Lorsque la session Spark est déjà initialisée, d'autres classes compilées et des JAR peuvent être importés en utilisant l' API spark.addArtifact().

remarque

Lors de l'upload de JARs, tous les fichiers JAR de dépendance transitive doivent être inclus pour l'upload. Les APIs n'effectuent aucune détection automatique des dépendances transitives.

UDFs avec des dépendances tierces​

Si vous avez ajouté une dépendance Maven dans build.sbt qui est utilisée dans une UDF mais qui n'est pas disponible sur le cluster Databricks, par exemple :

// In build.sbt
libraryDependencies += "org.apache.commons" % "commons-text" % "1.10.0"
Scala
// In your code
import org.apache.commons.text.StringEscapeUtils

// ClassNotFoundException thrown during UDF execution of this function on the server side
val escapeUdf = udf((text: String) => {
  StringEscapeUtils.escapeHtml4(text)
})

Utilisez spark.addArtifact() avec ivy:// pour download les dépendances de Maven :

  1. Ajoutez la bibliothèque oro à votre fichier build.sbt

    libraryDependencies ++= Seq(
      "org.apache.commons" % "commons-text" % "1.10.0" % Provided,
      "oro" % "oro" % "2.0.8"  // Required for ivy:// to work
    )

  2. Ajoutez l'artefact après avoir créé la session avec l'API addArtifact() :

    Scala
    def getSession(): SparkSession = {
      if (sys.env.contains("DATABRICKS_RUNTIME_VERSION")) {
        SparkSession.active
      } else {
        val spark = DatabricksSession.builder()
          .addCompiledArtifacts(Main.getClass.getProtectionDomain.getCodeSource.getLocation.toURI)
          .getOrCreate()

        // Convert Maven coordinates to ivy:// format
        // From: "org.apache.commons" % "commons-text" % "1.10.0"
        // To:   ivy://org.apache.commons:commons-text:1.10.0
        spark.addArtifact("ivy://org.apache.commons:commons-text:1.10.0")

        spark
      }
    }

APIs de Dataset typées​

Les APIs Dataset typés permettent d'effectuer des Transformations telles que map(), filter(), mapPartitions() et des agrégations sur les datasets résultants. L'upload de la classe compilée et des JAR vers le cluster à l'aide de l'API addCompiledArtifacts() s'applique également à ceux-ci, votre code doit donc se comporter différemment selon l'endroit où il s'exécute :

  • Développement local avec Databricks Connect : upload les artefacts vers le cluster distant.
  • Déployé sur Databricks s'exécutant sur le cluster : Pas besoin d'upload quoi que ce soit car les classes y sont déjà.

L'application Scala suivante utilise l'API map() pour modifier un nombre dans une colonne de résultats en une chaîne préfixée.

Scala
import com.databricks.connect.DatabricksSession
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions.{col, udf}

object Main {
  def main(args: Array[String]): Unit = {
    val sourceLocation = getClass.getProtectionDomain.getCodeSource.getLocation.toURI

    val spark = DatabricksSession.builder()
      .addCompiledArtifacts(sourceLocation)
      .getOrCreate()

    spark.range(3).map(f => s"row-$f").show()
  }
}

Dépendances JAR externes​

Si vous utilisez une bibliothèque privée ou tierce qui n'est pas sur le cluster :

Scala
import com.mycompany.privatelib.DataProcessor

// ClassNotFoundException thrown during UDF execution of this function on the server side
val myUdf = udf((data: String) => {
  DataProcessor.process(data)
})

Upload les JAR externes depuis votre dossier lib/ lors de la création de la session :

Scala
def getSession(): SparkSession = {
  if (sys.env.contains("DATABRICKS_RUNTIME_VERSION")) {
    SparkSession.active
  } else {
    val builder = DatabricksSession.builder()
      .addCompiledArtifacts(Main.getClass.getProtectionDomain.getCodeSource.getLocation.toURI)

     // Add all JARs from lib/ folder
     val libFolder = new java.io.File("lib")
     builder.addCompiledArtifacts(libFolder.toURI)


   builder.getOrCreate()
  }
}

Ceci importe automatiquement tous les fichiers JAR de votre répertoire lib/ vers Databricks lors de l'exécution locale.

Projets comportant plusieurs modules​

Dans un projet SBT à plusieurs modules, getClass.getProtectionDomain.getCodeSource.getLocation.toURI renvoie uniquement l'emplacement du module actuel. Si votre UDF utilise des classes d'autres modules, vous obtiendrez ClassNotFoundException.

my-project/
├── module-a/  (main application)
├── module-b/  (utilities - module-a depends on this)

Utilisez getClass d'une classe dans chaque module pour obtenir tous leurs emplacements et les upload séparément :

Scala
// In module-a/src/main/scala/Main.scala
import com.company.moduleb.DataProcessor  // From module-b

def getSession(): SparkSession = {
  if (sys.env.contains("DATABRICKS_RUNTIME_VERSION")) {
    SparkSession.active
  } else {
    // Get location using a class FROM module-a
    val moduleALocation = Main.getClass
      .getProtectionDomain.getCodeSource.getLocation.toURI

    // Get location using a class FROM module-b
    val moduleBLocation = DataProcessor.getClass
      .getProtectionDomain.getCodeSource.getLocation.toURI

    DatabricksSession.builder()
      .addCompiledArtifacts(moduleALocation)  // Upload module-a
      .addCompiledArtifacts(moduleBLocation)  // Upload module-b
      .getOrCreate()
  }
}

Limitations​

  • Le support des UDF sur le compute serverless suit toujours la version mineure correspondante initiale de Databricks Connect. Pour les versions prises en charge, consultez la table de compatibilité des versions.
Sur cette page