Snowpark Connect for Spark properties¶

Comments¶

Configuration that enables Snowflake Common Table Expressions (CTEs). This configuration optimizes the Snowflake queries in which there are a lot of repetitive code blocks. This modification will lead to improvements in both query compilation and execution performance.

Comments¶

When set to true, changes the behavior of certain operations:

• bit_get/getbit — Explicit use of Snowflake getbit function

• hash — Explicit use of Snowflake hash function

• md5 — Explicit use of Snowflake md5 function

• Renaming table columns — Allows for altering table columns

Comments¶

By default, Snowpark Connect for Spark treats all integral types as Long types. This is caused by the way numbers are represented in Snowflake. Integral types emulation allows for an exact mapping between Snowpark and Spark types when reading from datasources.

The default option client_default activates the emulation only when the script is executed from the Scala client. Integral types are mapped based on the following precisions:

When other precisions are found, the final type is mapped to the DecimalType.

Comments¶

To read only an exact subset of partitioned files from the provided directory, additional configuration is required. This feature is only available for files stored on external stages. To prune the read files, Snowpark Connect for Spark uses external tables.

This feature is enabled when the configuration snowpark.connect.sql.partition.external_table_location is set. It should contain existing database and schema names where external tables will be created.

Reading parquet files that are stored on external stages will create an external table; for files on internal stages, it will not be created. Providing the schema will reduce the execution time, eliminating the cost of inferring it from sources.

For best performance, filter according to the Snowflake External Tables filtering limitations.

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

Comments¶

This configuration works very similarly to the snowpark.connect.udf.python.imports. With it, you can specify external libraries and files for Java UDFs created using registerJavaFunction. Configurations are mutually exclusive to prevent unnecessary dependency mixing.

To include external libraries and files, you provide stage paths to the files as the value of the configuration setting snowpark.connect.udf.java.imports. The configuration value should be an array of stage paths to the files, where the paths are separated by commas.

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

Comments¶

You can use this to define additional packages to be available in Python UDFs. The value is a comma-separated list of dependencies.

You can discover the list of supported packages by executing the following SQL in 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()

Comments¶

This property determines whether UDTFs should use Spark-compatible behavior or the default Snowpark behavior. When set to true, it applies a compatibility wrapper that mimics Spark’s output type coercion and error handling patterns.

When enabled, UDTFs use a compatibility wrapper that applies Spark-style automatic type coercion (e.g., string “true” to boolean, boolean to integer) and error handling. The wrapper also converts table arguments to Row-like objects for both positional and named access, and properly handles SQL null values to match Spark’s behavior patterns.

Comments¶

Snowflake does not support duplicate column names.

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'

Comments¶

This configuration works very similarly to the snowpark.connect.udf.python.imports. You can use it to specify external libraries and files for Java UDFs created using registerJavaFunction. Configurations are mutually exclusive to prevent unnecessary dependency mixing.

To include external libraries and files, you provide stage paths to the files as the value of the configuration setting snowpark.connect.udf.java.imports. The value is an array of stage paths to the files, where the paths are separated by commas.

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

Comments¶

Configuration allows for defining additional packages available in Python UDFs. The value is a comma separated list of dependencies.

You can discover the list of supported packages by executing the following SQL in 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()

Comments¶

This configuration determines whether UDTFs should use Spark-compatible behavior or the default Snowpark behavior. When enabled (true), it applies a compatibility wrapper that mimics Spark’s output type coercion (for example, string “true” to boolean, boolean to integer) and error handling patterns.

The wrapper also converts table arguments to row-like objects for both positional and named access, and properly handles SQL null values to match Spark’s behavior patterns.

Comments¶

Snowflake tables do not support user-defined partitioning, and by default, partition overwrites will result in an error. Enabling this option allows using INSERT OVERWRITE <table> PARTITION(<partition spec>) syntax to perform overwrites.

The <partition spec> will accept any columns that exist in the target table.

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

Comments¶

By default, Snowpark Connect for Spark resolves Python packages from Snowflake’s curated Anaconda channel. Setting this configuration to an artifact repository name allows resolving packages from PyPI or other configured sources, enabling the use of packages that are not available in the Anaconda channel.

For information on how to create and configure an artifact repository in Snowflake, see Using third-party packages.

Changing this configuration invalidates cached UDFs and UDTFs, causing them to be recreated with the new repository on next invocation.

This configuration applies to the following operations:

• UDFs registered via @udf decorator or spark.udf.register()

• UDTFs registered via @udtf decorator or 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()

Comments¶

Some third-party Python packages (such as TensorFlow, XGBoost, and certain scientific libraries) are built only for the x86 CPU architecture. Setting this configuration to x86 adds RESOURCE_CONSTRAINT=(architecture='x86') to the CREATE FUNCTION statement generated by Snowpark Connect for Spark, ensuring the UDF runs on x86-compatible infrastructure.

To use this configuration, you must execute your workload on a warehouse that has been created with an x86 resource constraint. The following resource constraint values support x86:

• MEMORY_1X_x86 (minimum warehouse size: XSMALL)

• MEMORY_16X_x86 (minimum warehouse size: MEDIUM)

• MEMORY_64X_x86 (minimum warehouse size: LARGE)

If the warehouse does not have an x86 resource constraint, UDF execution will fail.

This configuration applies to the following operations:

• UDFs registered via @udf decorator or spark.udf.register()

• UDTFs registered via @udtf decorator or 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()