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’elles sont activées (true), cela applique un wrapper de compatibilité qui imite la coercition des types de sortie de Spark (par exemple, la chaîne « true » en booléen, le booléen en entier) et les modèles de gestion des erreurs.

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¶

Les tables Snowflake ne prennent pas en charge le partitionnement défini par l’utilisateur et, par défaut, les écrasements de partition entraîneront une erreur. L’activation de cette option permet d’utiliser la syntaxe INSERT OVERWRITE <table> PARTITION(<partition spec>) pour effectuer des écrasements.

Le <partition spec> acceptera toutes les colonnes qui existent dans la table cible.

spark.conf.set("snowpark.connect.sql.emulatePartitionOverwritesForSnowflakeTables", True)

# create the students and persons tables as standard Snowflake tables
students_data = [
  ("Ashua Hill", "456 Erica Ct, Cupertino", 111111),
  ("Brian Reed", "723 Kern Ave, Palo Alto", 222222)
]

students_df = spark.createDataFrame(students_data, ["name", "address", "student_id"])
students_df.write.mode("overwrite").saveAsTable("students")

persons_data = [
    ("Dora Williams", "134 Forest Ave, Menlo Park", 123456789),
    ("Eddie Davis", "245 Market St, Milpitas", 345678901)
]

persons_df = spark.createDataFrame(persons_data, ["name", "address", "ssn"])
persons_df.write.mode("overwrite").saveAsTable("persons")

# overwrites all rows in the students table that have a student_id of 222222
spark.sql("""
    INSERT OVERWRITE students PARTITION (student_id = 222222)
    SELECT name, address FROM persons WHERE name = 'Dora Williams'
""").collect()

Commentaires¶

Par défaut, Snowpark Connect for Spark résout les paquets Python du canal Anaconda de Snowflake. La définition de cette configuration sur un nom de dépôt d’artefacts permet de résoudre les paquets de PyPI ou d’autres sources configurées, afin d’utiliser des paquets qui ne sont pas disponibles dans le canal Anaconda.

Pour plus d’informations sur la création et la configuration d’un dépôt d’artefacts dans Snowflake, consultez Utilisation de paquets tiers.

La modification de cette configuration invalide les UDFs et les UDTFs mises en cache, ce qui entraîne leur recréation avec le nouveau dépôt lors de l’appel suivant.

Cette configuration s’applique aux opérations suivantes :

• UDFs enregistrée via le décorateur @udf ou spark.udf.register()

• UDTFs enregistrée via le décorateur @udtf ou spark.udtf.register()

• applyInPandas via groupBy().applyInPandas()

• mapInArrow via DataFrame.mapInArrow()

• cogroup via groupBy().cogroup().applyInPandas()

spark.conf.set("snowpark.connect.artifact_repository", "my_pypi_repo")
spark.conf.set("snowpark.connect.udf.packages", "[pykalman]")

@udf(returnType=DoubleType())
def kalman_smooth_value(value: float) -> float:
    import numpy as np
    from pykalman import KalmanFilter

    kf = KalmanFilter(
        transition_matrices=[1],
        observation_matrices=[1],
        initial_state_mean=0,
        initial_state_covariance=1,
        observation_covariance=1,
        transition_covariance=0.1,
    )
    observations = np.array([value, value, value])
    smoothed_state_means, _ = kf.smooth(observations)
    return float(smoothed_state_means[-1][0])

df = spark.createDataFrame([(1, 10.0), (2, 20.0), (3, 30.0)], ["id", "value"])
df.select("id", kalman_smooth_value("value").alias("smoothed")).show()

Commentaires¶

Certains paquets Python de tiers (comme TensorFlow,XGBoost et certaines bibliothèques scientifiques) sont conçus uniquement pour l’architecture CPU x86. Définir cette configuration sur x86 ajoute:code:RESOURCE_CONSTRAINT=(architecture='x86') à l’instruction CREATE FUNCTION générée par|spconnect|, garantissant ainsi que la UDF s’exécute sur une infrastructure x86 compatible.

Pour utiliser cette configuration, vous devez exécuter votre charge de travail sur un entrepôt qui a été créé avec une contrainte de ressources x86. Les valeurs de contrainte de ressources suivantes prennent en charge x86 :

• MEMORY_1X_x86 (taille minimale d’entrepôt : XSMALL)

• MEMORY_16X_x86 (taille minimale d’entrepôt : MEDIUM)

• MEMORY_64X_x86 (taille minimale d’entrepôt : LARGE)

Si l’entrepôt n’a pas de contrainte de ressources x86, l’exécution de la UDF échouera.

Cette configuration s’applique aux opérations suivantes :

• UDFs enregistrée via le décorateur @udf ou spark.udf.register()

• UDTFs enregistrée via le décorateur @udtf ou spark.udtf.register()

• applyInPandas via groupBy().applyInPandas()

CREATE WAREHOUSE my_x86_warehouse WITH
  WAREHOUSE_SIZE = 'MEDIUM'
  WAREHOUSE_TYPE = 'SNOWPARK-OPTIMIZED'
  RESOURCE_CONSTRAINT = 'MEMORY_16X_x86';

USE WAREHOUSE my_x86_warehouse;

spark.conf.set("snowpark.connect.udf.resource_constraint.architecture", "x86")

@udf(returnType=IntegerType())
def add_one(x: int) -> int:
    return x + 1

df = spark.createDataFrame([(1,), (2,), (3,)], ["value"])
df.select(add_one(df["value"]).alias("result")).show()