Snowpark Connect for Spark-Kompatibilitätshandbuch¶

Nicht unterstützte Datentypen¶

• DayTimeIntervalType

• YearMonthIntervalType

• UserDefinedTypes

Implizite Datentypkonvertierung¶

Beachten Sie bei Verwendung von Snowpark Connect for Spark, wie mit Datentypen umgegangen werden soll. Snowpark Connect for Spark steht implizit für ByteType, ShortType`und :code:`IntegerType als LongType. Das bedeutet, dass Sie zwar Spalten oder Daten mit ByteType, ShortType oder IntegerType definieren können, die Daten aber durch Snowpark Connect for Spark als LongType dargestellt und zurückgegeben werden. Ebenso kann eine implizite Konvertierung auch für FloatType und DoubleType abhängig von den spezifischen Operationen und dem Kontext erfolgen. Das Snowflake-Ausführungsmodul übernimmt intern die Datentypkomprimierung und kann die Daten tatsächlich als Byte oder Short speichern. Diese gelten jedoch als Implementierungsdetails und sind für den Endbenutzer nicht zugänglich.

Semantisch hat diese Darstellung keinen Einfluss auf die Korrektheit Ihrer Spark-Abfragen.

Das folgende Beispiel zeigt einen Unterschied bei der Behandlung von Datentypen in Abfrageergebnissen zwischen Spark und Snowpark Connect for Spark.

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)

NullType-Nuance¶

Snowpark Connect for Spark unterstützt nicht den Datentyp NullType, der in Spark unterstützt wird. Dies führt zu Verhaltensänderungen bei der Verwendung von Null oder None in Datenframes.

In Spark wird ein Literal NULL (zum Beispiel mit lit(None)) automatisch als NullType abgeleitet. In Snowpark Connect for Spark wird es während der Schema-Inferenz als StringType abgeleitet.

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)

Strukturierte Datentypen in ArrayType, MapType und ObjectType¶

Die Unterstützung strukturierter Typen ist zwar standardmäßig in Snowpark Connect for Spark nicht verfügbar, aber ARRAY-, MAP- und Object-Datentypen werden als generische, nicht typisierte Sammlungen behandelt. Dies bedeutet, dass Elementtypen, Feldnamen, Schemas oder Null-Zulässigkeiten nicht erzwungen werden, im Gegensatz zu dem, was die Unterstützung strukturierter Typen bieten würde.

Wenn Sie von diesem Support abhängig sind, arbeiten Sie mit Ihrem Kundenbetreuer-Team zusammen, um dieses Feature für Ihr Konto zu aktivieren.

StructType-Unterschiede¶

Wenn Spark einen StructType zur Verwendung in einer benutzerdefinierten Funktion (UDF) konvertiert, wird er in einen tuple -Typ in Python umgewandelt. Snowpark Connect for Spark konvertiert einen StructType in einen dict-Typ in Python. Es gibt grundlegende Unterschiede in Bezug auf den Elementzugriff und die Ausgabe.

• Spark greift auf Indizes mit 0, 1, 2, 3 usw. zu.

• Snowpark Connect for Spark greift mit „_1“, „_2“ usw. auf Indizes zu.

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}

Iteratortyp in UDFs¶

Der Iterator wird nicht als Rückgabetyp oder als Eingabetyp unterstützt.

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

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

Importieren von Dateien in eine Python-UDF¶

Mit Snowpark Connect for Spark können Sie externe Bibliotheken und Dateien in Python-UDFs angeben. Snowflake schließt Python-Dateien und -Archive in den Ausführungskontext Ihres Codes ein. Sie können Funktionen aus diesen enthaltenen Dateien ohne zusätzliche Schritte in eine UDF importieren. Dieses Verhalten bei der Behandlung von Abhängigkeiten funktioniert wie unter Erstellen eines Python-UDF mit aus einem Stagingbereich hochgeladenem Code beschrieben.

Um externe Bibliotheken und Dateien einzubeziehen, geben Sie Stagingbereichspfade zu den Dateien als Wert der Konfigurationseinstellung snowpark.connect.udf.imports an. Der Wert der Konfiguration sollte ein Array von Stagingbereichspfaden zu den Dateien sein, wobei die Pfade durch Kommas getrennt sind.

Der Code im folgenden Beispiel enthält zwei Dateien im UDF-Ausführungskontext. Die UDF importiert Funktionen aus diesen Dateien und verwendet sie in ihrer Logik.

# 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()

Sie können die snowpark.connect.udf.imports-Einstellung verwenden, um auch andere Arten von Dateien einzuschließen, z. B. Dateien mit Daten, die Ihr Code lesen muss. Beachten Sie, dass Ihr Code in diesem Fall nur aus den eingebundenen Dateien lesen sollte. Alle Schreibvorgänge in solche Dateien gehen nach Beendigung der Ausführung der Funktion verloren.

# 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()

Unterstützung von Snowflake Horizon Catalog-Anbietern¶

• Nur Snowflake wird als Kataloganbieter unterstützt.

Nicht unterstützte Katalog-APIs¶

• registerFunction

• listFunctions

• getFunction

• functionExists

• createExternalTable

Teilweise unterstützte Katalog-APIs¶

• createTable (keine Unterstützung externer Tabellen)

Snowflake-verwaltete Iceberg-Tabelle¶

Snowpark Connect für Spark funktioniert mit Apache Iceberg™-Tabellen, einschließlich extern verwalteter Iceberg-Tabellen und katalogverknüpfter Datenbanken.

Externe verwaltete Iceberg-Tabelle¶