Propriétés Snowpark Connect for Spark¶

Commentaires¶

Configuration activant les expressions de table communes (CTEs) Snowflake. Cette configuration optimise les requêtes Snowflake dans lesquelles il existe beaucoup de blocs de code répétitifs. Cette modification conduira à l’amélioration des performances de compilation et d’exécution des requêtes.

Commentaires¶

Lorsque défini sur true, modifie le comportement de certaines opérations :

• bit_get/getbit : utilisation explicite de la fonction Getbit de Snowflake

• hash : utilisation explicite de la fonction de hachage de Snowflake

• md5 : utilisation explicite de la fonction md5 de Snowflake

• Renommer les colonnes de la table : permet de modifier les colonnes de la table

Commentaires¶

Par défaut, Snowpark Connect for Spark traite tous les types intégraux comme des types Long. Cela est dû à la manière dont les nombres sont représentés dans Snowflake. L’émulation des types intégraux permet un mappage exact entre les types Snowpark et Spark lors de la lecture à partir de sources de données.

L’option client_default par défaut active l’émulation uniquement lorsque le script est exécuté à partir du client Scala. Les types intégraux sont mappés en fonction des précisions suivantes :

Lorsque d’autres précisions sont trouvées, le type final est mappé vers le DecimalType.

Commentaires¶

Pour lire uniquement un sous-ensemble exact de fichiers partitionnés à partir du répertoire fourni, une configuration supplémentaire est nécessaire. Cette fonctionnalité n’est disponible que pour les fichiers stockés sur des zones de préparation externes. Pour nettoyer les fichiers lus, Snowpark Connect for Spark utilise des tables externes.

Cette fonctionnalité est activée lorsque la configuration snowpark.connect.sql.partition.external_table_location est définie. Elle doit contenir des noms de bases de données et de schémas existants dans lesquels des tables externes seront créées.

La lecture de fichiers Parquet stockés dans des zones de préparation externes entraînera la création d’une table externe. Pour les fichiers dans des zones de préparation internes, elle ne sera pas créée. Fournir le schéma réduira le temps d’exécution, en éliminant le coût de l’inférence à partir des sources.

Pour des performances optimales, filtrez en fonction des limitations de filtrage des tables externes Snowflake.

spark.conf.set("snowpark.connect.sql.partition.external_table_location", "<database-name>.<schema-name>")

spark.read.parquet("@external-stage/example").filter(col("x") > lit(1)).show()

schema = StructType([StructField("x",IntegerType()),StructField("y",DoubleType())])

spark.read.schema(schema).parquet("@external-stage/example").filter(col("x") > lit(1)).show()

Commentaires¶

Cette configuration fonctionne de manière très similaire à snowpark.connect.udf.python.imports. Avec elle, vous pouvez spécifier des bibliothèques externes et des fichiers pour les UDFs Java créées à l’aide de `registerJavaFunction<https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.UDFRegistration.registerJavaFunction.html>`_. Les configurations s’excluent mutuellement pour empêcher les mélanges inutiles de dépendances.

Pour inclure des bibliothèques et des fichiers externes, vous fournissez des chemins d’accès aux zones de préparation aux fichiers en tant que valeur du paramètre de configuration snowpark.connect.udf.java.imports. La valeur de la configuration doit être un tableau de chemins d’accès aux zones de préparation, où les chemins sont séparés par des virgules.

# Files need to be previously staged

spark.conf.set("snowpark.connect.udf.java.imports", "[@stage/library.jar]")

spark.registerJavaFunction("javaFunction", "com.example.ExampleFunction")

spark.sql("SELECT javaFunction('arg')").show()

Commentaires¶

Vous pouvez l’utiliser pour définir des paquets supplémentaires à utiliser dans les UDFs Python. La valeur est une liste de dépendances séparées par des virgules.

Vous pouvez découvrir la liste des paquets pris en charge en exécutant la commande SQL suivante dans Snowflake :

SELECT * FROM INFORMATION_SCHEMA.PACKAGES WHERE LANGUAGE = 'python';

spark.conf.set("snowpark.connect.udf.packages", "[numpy]")

@udtf(returnType="val: int")

class Powers:

  def eval(self, x: int):
      import numpy as np

      for v in np.power(np.array([x, x, x]), [0, 1, 2]):
          yield (int(v),)

spark.udtf.register(name="powers", f=Powers)

spark.sql("SELECT * FROM powers(10)").show()

Commentaires¶

Cette propriété détermine si les UDTFs doivent utiliser un comportement compatible avec Spark ou le comportement par défaut de Snowpark. Lorsqu’elle est définie sur true, elle applique un wrapper de compatibilité qui imite les modèles de coercition des types de sortie et de traitement des erreurs de Spark.

Lorsqu’elles sont activées, les UDTFs utilisent un wrapper de compatibilité qui applique une coercition des types automatique (par exemple, chaîne « true » en booléen, booléen en entier) et un traitement des erreurs à la manière de Spark. Le wrapper convertit également les arguments de table en objets de type ligne pour l’accès positionnel et nommé, et traite correctement les valeurs SQL « null » pour correspondre aux schémas de comportement de Spark.

Commentaires¶

Snowflake ne prend pas en charge les noms de colonne dupliqués.

df = spark.createDataFrame([
(1, 1),
(2, 2)
], ["foo", "foo"])

df.show() # works

df.createTempView("df_view") # Fails with SQL compilation error: duplicate column name 'foo'

Commentaires¶

Cette configuration fonctionne de manière très similaire à snowpark.connect.udf.python.imports. Vous pouvez l’utiliser pour spécifier des bibliothèques externes et des fichiers pour les UDFs Java créées à l’aide de `registerJavaFunction<https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.UDFRegistration.registerJavaFunction.html>`_. Les configurations s’excluent mutuellement pour empêcher les mélanges inutiles de dépendances.

Pour inclure des bibliothèques et des fichiers externes, vous fournissez des chemins d’accès aux zones de préparation aux fichiers en tant que valeur du paramètre de configuration snowpark.connect.udf.java.imports. La valeur est un tableau de chemins d’accès aux fichiers, où les chemins sont séparés par des virgules.

# Files need to be previously staged

spark.conf.set("snowpark.connect.udf.java.imports", "[@stage/library.jar]")

spark.registerJavaFunction("javaFunction", "com.example.ExampleFunction")

spark.sql("SELECT javaFunction('arg')").show()

Commentaires¶

La configuration permet de définir des paquets supplémentaires disponibles dans les UDFs Python. La valeur est une liste de dépendances séparées par des virgules.

Vous pouvez découvrir la liste des paquets pris en charge en exécutant la commande SQL suivante dans Snowflake :

SELECT * FROM INFORMATION_SCHEMA.PACKAGES WHERE LANGUAGE = 'python';

spark.conf.set("snowpark.connect.udf.packages", "[numpy]")

@udtf(returnType="val: int")

class Powers:

  def eval(self, x: int):
      import numpy as np

      for v in np.power(np.array([x, x, x]), [0, 1, 2]):
          yield (int(v),)

spark.udtf.register(name="powers", f=Powers)

spark.sql("SELECT * FROM powers(10)").show()

Commentaires¶

Cette configuration détermine si les UDTFs doivent utiliser un comportement compatible avec Spark ou le comportement par défaut de Snowpark. Lorsqu’elle est activée (true), elle applique un wrapper de compatibilité qui imite les modèles de coercition des types de sortie et de traitement des erreurs de Spark.

Lorsqu’elles sont activées, les UDTFs utilisent un wrapper de compatibilité qui applique une coercition des types automatique (par exemple, chaîne « true » en booléen, booléen en entier) et un traitement des erreurs à la manière de Spark. Le wrapper convertit également les arguments de table en objets de type ligne pour l’accès positionnel et nommé, et traite correctement les valeurs SQL « null » pour correspondre aux schémas de comportement de Spark.