Propriedades do Snowpark Connect for Spark¶

Comentários¶

Configuração que habilita as Expressões de Tabela Comuns do Snowflake (CTEs). Essa configuração otimiza as consultas do Snowflake nas quais há muitos blocos de código repetitivos. Essa modificação vai melhorar tanto a compilação de consultas quanto o desempenho da execução.

Comentários¶

Quando definida como true, altera o comportamento de determinadas operações:

• bit_get/getbit: uso explícito da função getbit do Snowflake

• hash: uso explícito da função de hash do Snowflake

• md5: uso explícito da função md5 do Snowflake

• Renomear colunas de tabela: permite alterar as colunas de tabela

Comentários¶

Por padrão, o Snowpark Connect for Spark trata todos os tipos integrais como Long. Isso é causado pela maneira como os números são representados no Snowflake. A emulação de tipos integrais permite um mapeamento exato entre os tipos do Snowpark e do Spark ao ler fontes de dados.

A opção padrão client_default ativa a emulação somente quando o script é executado do cliente Scala. Os tipos integrais são mapeados com base nas seguintes precisões:

Quando outras precisões são encontradas, o tipo final é mapeado para o DecimalType.

Comentários¶

Para ler apenas um subconjunto exato de arquivos particionados do diretório fornecido, é necessária uma configuração adicional. Este recurso só está disponível para arquivos armazenados em áreas de preparação externas. Para remover os arquivos de leitura, o Snowpark Connect for Spark usa tabelas externas.

Este recurso é habilitado quando a configuração snowpark.connect.sql.partition.external_table_location é definida. Ele deve conter os nomes existentes de bancos de dados e esquemas em que as tabelas externas serão criadas.

A leitura de arquivos Parquet armazenados em áreas de preparação externas cria uma tabela externa. Para arquivos em áreas de preparação internas, ela não será criada. O fornecimento do esquema reduz o tempo de execução, o que elimina o custo para inferi-lo com base nas fontes.

Para melhor desempenho, filtre de acordo com as limitações de filtragem de tabelas externas do Snowflake.

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

Comentários¶

Essa configuração funciona de forma muito semelhante a snowpark.connect.udf.python.imports. Com ela, você pode especificar bibliotecas e arquivos externos para UDFs Java criadas usando registerJavaFunction. As configurações são mutuamente exclusivas para impedir uma mistura de dependências desnecessária.

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.java.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.

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

Comentários¶

Você pode usá-la para definir pacotes adicionais a serem disponibilizados em UDFs Python. O valor é uma lista de dependências separadas por vírgulas.

Você pode descobrir a lista de pacotes compatíveis executando o seguinte SQL no 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()

Comentários¶

Esta propriedade determina se as UDTFs devem usar o comportamento compatível com Spark ou o comportamento padrão do Snowpark. Quando definida como true, aplica um wrapper de compatibilidade que imita a coerção de tipo de saída e os padrões de tratamento de erros do Spark.

Quando habilitada, as UDTFs usam um wrapper de compatibilidade que aplica a coerção automática de tipo (por exemplo, cadeia de caracteres «true» para booliana, booliana para número inteiro) e tratamento de erros no estilo do Spark. O wrapper também converte argumentos de tabela em objetos semelhantes à linha para acesso posicional e nomeado, além de processar corretamente valores nulos SQL para corresponder aos padrões de comportamento do Spark.

Comentários¶

O Snowflake não oferece suporte a nomes de colunas duplicados.

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'

Comentários¶

Essa configuração funciona de forma muito semelhante a snowpark.connect.udf.python.imports. Você pode usá-la para especificar bibliotecas e arquivos externos para UDFs Java criadas usando registerJavaFunction. As configurações são mutuamente exclusivas para impedir uma mistura de dependências desnecessária.

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.java.imports. O valor é uma matriz de caminhos de área de preparação para os arquivos, com os caminhos separados por vírgulas.

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

Comentários¶

A configuração permite definir pacotes adicionais disponíveis nas UDFs Python. O valor é uma lista de dependências separadas por vírgulas.

Você pode descobrir a lista de pacotes compatíveis executando o seguinte SQL no 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()

Comentários¶

Esta configuração determina se as UDTFs devem usar o comportamento compatível com Spark ou o comportamento padrão do Snowpark. Quando habilitada (true), usa um wrapper de compatibilidade que aplica a coerção automática de tipo (por exemplo, cadeia de caracteres «true» para booliana, booliana para número inteiro) e tratamento de erros no estilo do Spark.

O wrapper também converte argumentos de tabela em objetos semelhantes à linha para acesso posicional e nomeado, além de processar corretamente valores nulos SQL para corresponder aos padrões de comportamento do Spark.

Comentários¶

As tabelas Snowflake não oferecem suporte a particionamento definido pelo usuário e, por padrão, a substituição de partições resultará em erro. A ativação dessa opção permite usar INSERT OVERWRITE <table> PARTITION(<partition spec>) para executar substituições.

<partition spec> aceitará todas as colunas existentes na tabela de destino.

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

Comentários¶

Por padrão, Snowpark Connect for Spark resolve pacotes Python do canal Anaconda selecionado do Snowflake. Definir essa configuração como um nome de repositório de artefatos permite a resolução de pacotes de PyPI ou outras fontes configuradas, possibilitando o uso de pacotes que não estão disponíveis no canal Anaconda.

Para obter informações sobre como criar e configurar um repositório de artefatos no Snowflake, consulte:doc:/developer-guide/udf/python/udf-python-packages .

Alterar essa configuração invalida UDFs e UDTFs em cache, fazendo com que elas sejam recriadas com o novo repositório na próxima invocação.

Essa configuração se aplica às seguintes operações:

• UDFs registradas via decorador @udf ou spark.udf.register()

• UDTFs registradas via decorador @udtf ou 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()

Comentários¶

Alguns pacotes Python de terceiros (como TensorFlow, XGBoost e certas bibliotecas científicas) são compilados apenas para a arquitetura de CPU x86. Definir essa configuração como x86 adiciona RESOURCE_CONSTRAINT=(architecture='x86') à instrução CREATE FUNCTION gerada por Snowpark Connect for Spark, garantindo que a UDF seja executada em uma infraestrutura compatível com x86.

Para usar essa configuração, execute sua carga de trabalho em um warehouse que tenha sido criado com uma restrição de recursos x86. Os seguintes valores de restrição de recursos são compatíveis com x86:

• MEMORY_1X_x86 (tamanho mínimo do warehouse: XSMALL)

• MEMORY_16X_x86 (tamanho mínimo do warehouse: MEDIUM)

• MEMORY_64X_x86 (tamanho mínimo do warehouse: LARGE)

Se o warehouse não tiver uma restrição de recurso x86, a execução da UDF falhará.

Essa configuração se aplica às seguintes operações:

• UDFs registradas via decorador @udf ou spark.udf.register()

• UDTFs registradas via decorador @udtf ou 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()