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

Importação de arquivos para UDF Python¶

Com o Snowpark Connect for Spark, você pode especificar bibliotecas e arquivos externos em UDFs Python. O Snowflake inclui arquivos Python e arquivamentos no contexto de execução do seu código. Você pode importar funções desses arquivos incluídos em uma UDF sem etapas adicionais. Esse comportamento de tratamento de dependência funciona conforme descrito em Como criar uma UDF Python com código carregado de um estágio.

Para incluir bibliotecas e arquivos externos, você fornece caminhos de área de preparação para os arquivos como valor da definição de configuração snowpark.connect.udf.imports. O valor da configuração deve ser uma matriz de caminhos de área de preparação para os arquivos, com os caminhos separados por vírgulas.

O código no exemplo a seguir inclui dois arquivos no contexto de execução da UDF. A UDF importa funções desses arquivos e as utiliza na lógica.

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

Você pode usar a configuração snowpark.connect.udf.imports para incluir também outros tipos de arquivos, como aqueles com dados que seu código precisa ler. Quando você faz isso, observe que seu código deve ler apenas os arquivos incluídos, qualquer gravação nesses arquivos será perdida após o término da execução da função.

# 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¶