Guide de compatibilité Snowpark Connect for Spark¶

Types de données non pris en charge¶

• DayTimeIntervalType

• YearMonthIntervalType

• UserDefinedTypes

Conversion implicite de type de données¶

Lorsque vous utilisez Snowpark Connect for Spark, gardez à l’esprit la façon dont les types de données sont traités. Snowpark Connect for Spark représente implicitement ByteType, ShortType et IntegerType comme LongType. Cela signifie que si vous pouvez définir des colonnes ou des données avec ByteType, ShortType ou IntegerType, les données seront représentées et renvoyées par Snowpark Connect for Spark comme LongType. De même, une conversion implicite peut également se produire pour FloatType et DoubleType en fonction des opérations et du contexte spécifiques. Le moteur d’exécution Snowflake gérera en interne la compression des types de données et peut stocker les données sous la forme Byte ou Short, mais ceux-ci sont considérés comme des détails de mise en œuvre et ne sont pas exposés à l’utilisateur final.

Sémantiquement, cette représentation n’aura pas d’impact sur l’exactitude de vos requêtes Spark.

L’exemple suivant montre une différence dans la manière dont Spark et Snowpark Connect for Spark gèrent les types de données dans les résultats des requêtes.

query = """
    SELECT * FROM VALUES
    (float(1.0), double(1.0), 1.0, "1", true, :code:`NULL`),
    (float(2.0), double(2.0), 2.0, "2", false, :code:`NULL`),
    (float(3.0), double(3.0), :code:`NULL`, "3", false, :code:`NULL`)
    AS tab(a, b, c, d, e, f)
    """

spark.sql(query).printSchema()

root
 |-- a: float (nullable = false)
 |-- b: double (nullable = false)
 |-- c: decimal(2,1) (nullable = true)
 |-- d: string (nullable = false)
 |-- e: boolean (nullable = false)
 |-- f: void (nullable = true)

snowpark_connect_spark.sql(query).printSchema()

root
 |-- a: double (nullable = false)
 |-- b: double (nullable = false)
 |-- c: decimal (nullable = true)
 |-- d: string (nullable = false)
 |-- e: boolean (nullable = true)
 |-- f: string (nullable = true)

Nuance NullType¶

Snowpark Connect for Spark ne prend pas en charge le type de données NullType, qui est un type de données pris en charge dans Spark. Cela entraîne des changements de comportement lors de l’utilisation de Null ou None dans les dataframes.

Dans Spark, un NULL littéral (par exemple, avec lit(None)) est automatiquement déduit comme un NullType. Dans Snowpark Connect for Spark, il est déduit comme un StringType lors de l’inférence de schéma.

df = self.spark.range(1).select(lit(None).alias("null_col"))
field = df.schema["null_col"]

# Spark: StructField('null_col', :code:`NullType`(), True)
# Snowpark Connect for Spark: StructField('null_col', :code:`StringType`(), True)

Types de données structurés dans ArrayType, MapType et ObjectType¶

Bien que la prise en charge des types structurés ne soit pas disponible par défaut dans Snowpark Connect for Spark, les types de données ARRAY, MAP et Object sont traités comme des collections génériques et non typées. Cela signifie qu’il n’y a pas d’application des types d’éléments, des noms de champs, du schéma ou de la nullabilité, contrairement à ce qui serait fourni par la prise en charge des types structurés.

Si vous dépendez de cette prise en charge, veuillez contacter votre équipe de compte afin d’activer cette fonctionnalité pour votre compte.

Différences StructType¶

Lorsque Spark convertit un StructType à utiliser dans une fonction définie par l’utilisateur (UDF), il le convertit en un type tuple en Python. Snowpark Connect for Spark convertira un StructType en un type dict en Python. Cela présente des différences fondamentales en matière de sortie et d’accès aux éléments.

• Spark accédera aux index avec 0, 1, 2, 3, et ainsi de suite.

• Snowpark Connect for Spark accédera aux index en utilisant “_1”, “_2”, etc.

def f(e):
    return e[0]

    df = self.spark.createDataFrame([((1.0, 1.0), (1, 1))], ["c1", "c2"])
    result = df.select("*", udf(f, DoubleType())("c1"))

# This results in an index access issue. Workaround is to use _1, _2 as indicies.
# Workaround:

def f(e):
    return e['_1']

row = (
    self.spark.range(1)
    .selectExpr("struct(1, 2) as struct")
    .select(
        udf(lambda x: x, "struct<col1:int,col2:int>")("struct"),
    )
    .first()
)

self.assertEquals(row[0], Row(col1=1, col2=2))

# Spark: Row(col1=1, col2=2)

# Snowpark Connect for Spark: {'col1': 1, 'col2': 2}

Type d’itérateur dans les UDFs¶

L’itérateur n’est pas pris en charge en tant que type de retour ou en tant que type d’entrée.

# This will not work
def func(iterator):
  for _ in iterator:
              ...

df = self.spark.range(10)
actual = df.repartition(1).mapInArrow(func, "a long").collect()

Importation de fichiers dans une UDF Python.¶

Avec Snowpark Connect for Spark, vous pouvez spécifier des bibliothèques externes et des fichiers dans les UDFs Python. Snowflake inclut des fichiers et des archives Python dans le contexte d’exécution de votre code. Vous pouvez importer des fonctions à partir de ces fichiers inclus dans une UDF sans étapes supplémentaires. Ce comportement de gestion des dépendances fonctionne comme décrit dans Création d’une UDF Python avec du code téléchargé à partir d’une zone de préparation.

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.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.

Le code de l’exemple suivant inclut deux fichiers dans le contexte d’exécution de l’UDF. L’UDF importe des fonctions de ces fichiers et les utilise dans sa logique.

# Files need to be previously staged
spark.conf.set("snowpark.connect.udf.imports", "[@stage/library.py, @other_lib.zip]")

@udf(returnType = StringType())
def import_example(input: str) -> str:
  from library import first_function
  from other_lib.custom import second_function

  return first_function(input) + second_function(input)

spark.range(1).select(import_read_example("example_string")).show()

Vous pouvez utiliser le paramètre snowpark.connect.udf.imports pour inclure d’autres types de fichiers, tels que ceux contenant des données que votre code doit lire. Notez que lorsque vous faites cela, votre code ne doit lire qu’à partir des fichiers inclus ; toutes les écritures dans ces fichiers seront perdues après la fin de l’exécution de la fonction.

# Files need to be previously staged
spark.conf.set("snowpark.connect.udf.imports", "[@stage/data.csv]")

@udf(returnType = StringType())
def import_read_example(file_name: str) -> str:
  with open(file_name) as f:
    return f.read()

spark.range(1).select(import_read_example("data.csv")).show()

Prise en charge du fournisseur de catalogue Snowflake Horizon¶

• Seul Snowflake est pris en charge en tant que fournisseur de catalogue.

APIs de catalogue non prises en charge¶

• registerFunction

• listFunctions

• getFunction

• functionExists

• createExternalTable

APIs de catalogue partiellement prises en charge¶

• createTable (pas de prise en charge des tables externes)

Table Iceberg gérée par Snowflake¶

Snowpark Connect pour Spark fonctionne avec les tables Apache Iceberg™, y compris les tables Iceberg gérées en externe et les bases de données liées à des catalogues.

Table Iceberg gérée en externe¶