Snowpark Connect for Spark-Eigenschaften¶

Kommentare¶

Konfiguration, die Allgemeine Snowflake-Tabellenausdrücke (CTEs) aktiviert. Diese Konfiguration optimiert die Snowflake-Abfragen, in denen es viele sich wiederholende Codeblöcke gibt. Diese Änderung wird zu Verbesserungen sowohl bei der Abfragekompilierung als auch bei der Ausführung führen.

Kommentare¶

Bei true ändert sich das Verhalten bestimmter Vorgänge:

• bit_get/getbit – Explizite Verwendung der Snowflake-GetBit-Funktion

• hash – Explizite Verwendung der Snowflake-Hash-Funktion

• md5 – Explizite Verwendung der Snowflake-md5-Funktion

• Umbenennen von Tabellenspalten – Ermöglicht das Ändern von Tabellenspalten

Kommentare¶

Standardmäßig behandelt Snowpark Connect for Spark alle ganzzahligen Typen als Long-Typen Dies wird durch die Art und Weise verursacht, in der Zahlen in Snowflake dargestellt werden. Die Emulierung von Ganzzahl-Typen ermöglicht beim Lesen von Datenquellen eine exakte Zuordnung zwischen Snowpark- und Spark-Typen.

Die Standardoption client_default aktiviert die Emulation nur, wenn das Skript vom Scala-Client ausgeführt wird. Ganzzahl-Typen werden auf der Grundlage der folgenden Genauigkeiten zugeordnet:

Wenn andere Genauigkeiten gefunden werden, wird der endgültige Typ dem DecimalType zugeordnet.

Kommentare¶

Um nur eine genaue Teilmenge der partitionierten Dateien aus dem angegebenen Verzeichnis zu lesen, ist eine zusätzliche Konfiguration erforderlich. Dieses Feature ist nur für Dateien verfügbar, die auf externen Stagingbereichen gespeichert sind. Um die gelesenen Dateien abzutrennen, verwendet Snowpark Connect for Spark externe Tabellen.

Dieses Feature wird aktiviert, wenn die Konfiguration snowpark.connect.sql.partition.external_table_location festgelegt ist. Sie sollte bestehende Datenbank- und Schemanamen enthalten, in denen externe Tabellen erstellt werden.

Beim Lesen von Parquet-Dateien, die in externen Stagingbereichen gespeichert sind, wird eine externe Tabelle erstellt. Für Dateien in internen Stagingbereichen wird er nicht erstellt. Die Bereitstellung des Schemas verkürzt die Ausführungszeit, sodass keine Kosten für die Ableitung aus den Quellen entfallen.

Um die beste Leistung zu erzielen, filtern Sie nach Einschränkungen beim Filtern von externen Snowflake-Tabellen.

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

Kommentare¶

Diese Konfiguration funktioniert sehr ähnlich wie snowpark.connect.udf.python.imports. Damit können Sie externe Bibliotheken und Dateien für Java-UDFs angeben, die mit registerJavaFunction erstellt wurden. Konfigurationen schließen sich gegenseitig aus, um eine unnötige Vermischung von Abhängigkeiten zu verhindern.

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

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

Kommentare¶

Sie können dies verwenden, um zusätzliche Pakete zu definieren, die in Python-UDFs verfügbar sein sollen. Der Wert ist eine durch Kommas getrennte Liste von Abhängigkeiten.

Sie können die Liste der unterstützten Pakete ermitteln, indem Sie in Snowflake den folgenden SQL-Befehl ausführen:

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

Kommentare¶

Diese Eigenschaft bestimmt, ob UDTFs Spark-kompatibles Verhalten oder Standard-Snowpark-Verhalten verwenden soll. Bei true wird ein Kompatibilitäts-Wrapper angewendet, der die Koersion des Ausgabetyps und die Fehlerbehandlungsmuster von Spark nachahmt.

Wenn aktiviert, verwenden UDTFs einen Kompatibilitäts-Wrapper, der die automatische Typenzwang (z. B. „Zeichenfolge ‚true‘“ auf „boolesch“, „boolesch“ auf „Ganzzahl“) und die Fehlerbehandlung anwendet. Der Wrapper konvertiert auch Tabellenargumente in zeilenähnliche Objekte für den Positions- und den Namenszugriff und behandelt korrekt SQL-Nullwerte entsprechend den Verhaltensmustern von Spark.

Kommentare¶

Snowflake unterstützt keine doppelten Spaltennamen.

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'

Kommentare¶

Diese Konfiguration funktioniert sehr ähnlich wie snowpark.connect.udf.python.imports. Sie können damit externe Bibliotheken und Dateien für Java-UDFs angeben, die mit registerJavaFunction erstellt wurden. Konfigurationen schließen sich gegenseitig aus, um eine unnötige Vermischung von Abhängigkeiten zu verhindern.

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

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

Kommentare¶

Die Konfiguration ermöglicht die Definition zusätzlicher Pakete, die in Python-UDFs verfügbar sind. Der Wert ist eine durch Kommas getrennte Liste von Abhängigkeiten.

Sie können die Liste der unterstützten Pakete ermitteln, indem Sie in Snowflake den folgenden SQL-Befehl ausführen:

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

Kommentare¶

Diese Konfiguration bestimmt, ob UDTFs Spark-kompatibles Verhalten oder Standard-Snowpark-Verhalten verwenden soll. Bei Aktivierung (true) wird ein Kompatibilitäts-Wrapper angewendet, der die Koersion des Ausgabetyps und die Fehlerbehandlungsmuster von Spark nachahmt.

Wenn aktiviert, verwenden UDTFs einen Kompatibilitäts-Wrapper, der die automatische Typenzwang (z. B. „Zeichenfolge ‚true‘“ auf „boolesch“, „boolesch“ auf „Ganzzahl“) und die Fehlerbehandlung anwendet. Der Wrapper konvertiert auch Tabellenargumente in zeilenähnliche Objekte für den Positions- und den Namenszugriff und behandelt korrekt SQL-Nullwerte entsprechend den Verhaltensmustern von Spark.