Guia de compatibilidade do Snowpark Connect for Spark¶

Tipos de dados sem suporte¶

• DayTimeIntervalType

• YearMonthIntervalType

• UserDefinedTypes

Conversão implícita do tipo de dados¶

Ao usar Snowpark Connect for Spark, tenha em mente como os tipos de dados são tratados. Snowpark Connect for Spark representa implicitamente ByteType, ShortType`e :code:`IntegerType como LongType. Isso significa que, embora você possa definir colunas ou dados com ByteType, ShortType`ou :code:`IntegerType, os dados serão representados e retornados por Snowpark Connect for Spark como LongType. Da mesma forma, a conversão implícita também pode ocorrer para FloatType e DoubleType dependendo das operações e do contexto específicos. O mecanismo de execução do Snowflake manipulará internamente a compactação do tipo de dados e poderá, de fato, armazenar os dados como Byte ou Short, mas esses são considerados detalhes de implementação e não são expostos ao usuário final.

Semanticamente, esta representação não afetará a exatidão de suas consultas do Spark.

O exemplo a seguir mostra uma diferença entre como o Spark e o Snowpark Connect for Spark manipulam tipos de dados em resultados de consulta.

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 de NullType¶

Snowpark Connect for Spark não oferece suporte ao tipo de dados NullType, que é um tipo de dados compatível no Spark. Isso causa mudanças de comportamento ao usar Null ou None em dataframes.

No Spark, um NULL literal (por exemplo, com lit(None)) é automaticamente inferido como um NullType. Em Snowpark Connect for Spark, ele é inferido como um StringType durante a inferência do esquema.

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)

Tipos de dados estruturados em ArrayType, MapType e ObjectType¶

Embora o suporte a tipos estruturados não esteja disponível por padrão em Snowpark Connect for Spark, os tipos de dados ARRAY, MAP e Object são tratados como coleções genéricas e sem tipo. Isso significa que não há aplicação de tipos de elementos, nomes de campos, esquema ou nulidade, ao contrário do que seria fornecido pelo suporte a tipos estruturados.

Se você tem uma dependência desse suporte, trabalhe com sua equipe de conta para habilitar esse recurso para sua conta.

Diferenças de StructType¶

Quando o Spark converte um StructType a ser usado em uma função definida pelo usuário (UDF), ele o converte em um tipo tuple em Python. Snowpark Connect for Spark converterá um StructType em um tipo dict em Python. Isso tem diferenças fundamentais no acesso e na saída do elemento.

• O Spark acessará índices com 0, 1, 2, 3 e assim por diante.

• Snowpark Connect for Spark acessará índices usando “_1”, “_2” e assim por diante.

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}

Tipo de iterador em UDFs¶

O iterador não é compatível como tipo de retorno ou como tipo de entrada.

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

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

Importing files to a Python UDF¶

With Snowpark Connect for Spark, you can specify external libraries and files in Python UDFs. Snowflake includes Python files and archives in your code’s execution context. You can import functions from these included files in a UDF without additional steps. This dependency-handling behavior works as described in Como criar uma UDF Python com código carregado de um estágio.

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

Code in the following example includes two files in the UDF’s execution context. The UDF imports functions from these files and uses them in its logic.

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

You can use the snowpark.connect.udf.imports setting to include other kinds of files as well, such as those with data your code needs to read. Note that when you do this, your code should only read from the included files; any writes to such files will be lost after the function’s execution ends.

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

Suporte ao provedor do Catálogo Snowflake Horizon¶

• Somente o Snowflake é compatível como provedor de catálogo.

APIs de catálogo sem suporte¶

• registerFunction

• listFunctions

• getFunction

• functionExists

• createExternalTable

APIs de catálogo parcialmente suportadas¶

• createTable (sem suporte para tabela externa)

Tabela Iceberg gerenciada pelo Snowflake¶

O Snowpark Connect para Spark funciona com tabelas Apache Iceberg™, incluindo tabelas Iceberg gerenciadas externamente e bancos de dados vinculados a catálogo.

Tabela Iceberg gerenciada externamente¶