Como trabalhar com DataFrames no Snowpark Python

No Snowpark, a principal forma de consultar e processar dados é através de um DataFrame. Este tópico explica como trabalhar com DataFrames.

Neste tópico:

Para obter e manipular dados, use a classe DataFrame. Um DataFrame representa um conjunto de dados relacionais que é avaliado lentamente: ele só é executado quando uma ação específica é acionada. Em certo sentido, um DataFrame é como uma consulta que precisa ser avaliada para obter dados.

Para obter dados em um DataFrame:

  1. Construa um DataFrame especificando a fonte dos dados para o conjunto de dados.

    Por exemplo, você pode criar um DataFrame para armazenar dados de uma tabela, um arquivo CSV externo, de dados locais ou a execução de uma instrução SQL.

  2. Especifique como o conjunto de dados no DataFrame deve ser transformado.

    Por exemplo, você pode especificar quais colunas devem ser selecionadas, como as linhas devem ser filtradas, como os resultados devem ser ordenados e agrupados, etc.

  3. Execute a instrução para obter os dados para o DataFrame.

    Para obter os dados no DataFrame, você deve invocar um método que execute uma ação (por exemplo, o método collect()).

As próximas seções explicam essas etapas com mais detalhes.

Como estabelecer os exemplos para esta seção

Alguns exemplos desta seção utilizam um DataFrame para consultar uma tabela chamada sample_product_data. Se você quiser executar esses exemplos, você pode criar essa tabela e preenchê-la com alguns dados executando as seguintes instruções SQL.

Você pode executar as instruções SQL usando o Snowpark Python:

session.sql('CREATE OR REPLACE TABLE sample_product_data (id INT, parent_id INT, category_id INT, name VARCHAR, serial_number VARCHAR, key INT, "3rd" INT)').collect()
Copy
[Row(status='Table SAMPLE_PRODUCT_DATA successfully created.')]

session.sql("""
INSERT INTO sample_product_data VALUES
(1, 0, 5, 'Product 1', 'prod-1', 1, 10),
(2, 1, 5, 'Product 1A', 'prod-1-A', 1, 20),
(3, 1, 5, 'Product 1B', 'prod-1-B', 1, 30),
(4, 0, 10, 'Product 2', 'prod-2', 2, 40),
(5, 4, 10, 'Product 2A', 'prod-2-A', 2, 50),
(6, 4, 10, 'Product 2B', 'prod-2-B', 2, 60),
(7, 0, 20, 'Product 3', 'prod-3', 3, 70),
(8, 7, 20, 'Product 3A', 'prod-3-A', 3, 80),
(9, 7, 20, 'Product 3B', 'prod-3-B', 3, 90),
(10, 0, 50, 'Product 4', 'prod-4', 4, 100),
(11, 10, 50, 'Product 4A', 'prod-4-A', 4, 100),
(12, 10, 50, 'Product 4B', 'prod-4-B', 4, 100)
""").collect()
Copy
[Row(number of rows inserted=12)]

Para verificar se a tabela foi criada, execute:

session.sql("SELECT count(*) FROM sample_product_data").collect()
Copy
[Row(COUNT(*)=12)]

Configuração dos exemplos em uma planilha Python

Para configurar e executar estes exemplos em uma planilha Python, crie a tabela de amostra e configure sua planilha Python.

  1. Crie uma planilha SQL e execute o seguinte:

CREATE OR REPLACE TABLE sample_product_data
  (id INT, parent_id INT, category_id INT, name VARCHAR, serial_number VARCHAR, key INT, "3rd" INT);

INSERT INTO sample_product_data VALUES
  (1, 0, 5, 'Product 1', 'prod-1', 1, 10),
  (2, 1, 5, 'Product 1A', 'prod-1-A', 1, 20),
  (3, 1, 5, 'Product 1B', 'prod-1-B', 1, 30),
  (4, 0, 10, 'Product 2', 'prod-2', 2, 40),
  (5, 4, 10, 'Product 2A', 'prod-2-A', 2, 50),
  (6, 4, 10, 'Product 2B', 'prod-2-B', 2, 60),
  (7, 0, 20, 'Product 3', 'prod-3', 3, 70),
  (8, 7, 20, 'Product 3A', 'prod-3-A', 3, 80),
  (9, 7, 20, 'Product 3B', 'prod-3-B', 3, 90),
  (10, 0, 50, 'Product 4', 'prod-4', 4, 100),
  (11, 10, 50, 'Product 4A', 'prod-4-A', 4, 100),
  (12, 10, 50, 'Product 4B', 'prod-4-B', 4, 100);

SELECT count(*) FROM sample_product_data;
Copy

  1. Crie uma planilha Python, definindo o mesmo contexto de banco de dados e esquema que a planilha SQL utilizada para criar a tabela sample_product_data.

Se você quiser usar os exemplos deste tópico em uma planilha Python, use o exemplo dentro da função do manipulador (por exemplo, main), e use o objeto Session que é passado para a função para criar DataFrames.

Por exemplo, chame o método table do objeto session para criar um DataFrame para uma tabela:

import snowflake.snowpark as snowpark
from snowflake.snowpark.functions import col

def main(session: snowpark.Session):
  df_table = session.table("sample_product_data")
Copy

Para revisar a saída produzida pela função, por exemplo, chamando o método show do objeto DataFrame, use a guia Output.

Para examinar o valor retornado pela função, escolha o tipo de dados do valor de retorno de Settings » Return type, e use a guia Results:

  • Se sua função retornar um DataFrame, use o tipo de retorno padrão de Table.

  • Se sua função retornar o list de Row do método collect de um objeto DataFrame, use Variant para o tipo de retorno.

  • Se sua função retornar qualquer outro valor que possa ser convertido em uma cadeia de caracteres, ou se sua função não retornar um valor, use String como o tipo de retorno.

Consulte Execução das planilhas Python para obter mais detalhes.

Como criar um DataFrame

Para criar um DataFrame, você pode usar os métodos e propriedades da classe Session. Cada um dos métodos a seguir constrói um DataFrame a partir de um tipo diferente de fonte de dados.

Você pode executar estes exemplos em seu ambiente de desenvolvimento local ou chamá-los dentro da função main definida em uma planilha Python.

  • Para criar um DataFrame a partir de dados em uma tabela, exibição ou fluxo, chame o método table:

    # Create a DataFrame from the data in the "sample_product_data" table.
df_table = session.table("sample_product_data")

# To print out the first 10 rows, call df_table.show()
    Copy

  • Para criar um DataFrame a partir de valores especificados, chame o método create_dataframe:

    # Create a DataFrame with one column named a from specified values.
df1 = session.create_dataframe([1, 2, 3, 4]).to_df("a")
df1.show()
# To return the DataFrame as a table in a Python worksheet use return instead of show()
# return df1
    Copy
    -------
|"A"  |
-------
|1    |
|2    |
|3    |
|4    |
-------

    Crie um DataFrame com 4 colunas, «a», «b», «c» e «d»:

    # Create a DataFrame with 4 columns, "a", "b", "c" and "d".
df2 = session.create_dataframe([[1, 2, 3, 4]], schema=["a", "b", "c", "d"])
df2.show()
# To return the DataFrame as a table in a Python worksheet use return instead of show()
# return df2
    Copy
    -------------------------
|"A"  |"B"  |"C"  |"D"  |
-------------------------
|1    |2    |3    |4    |
-------------------------

    Crie outro DataFrame com 4 colunas, «a», «b», «c» e «d»:

    # Create another DataFrame with 4 columns, "a", "b", "c" and "d".
from snowflake.snowpark import Row
df3 = session.create_dataframe([Row(a=1, b=2, c=3, d=4)])
df3.show()
# To return the DataFrame as a table in a Python worksheet use return instead of show()
# return df3
    Copy
    -------------------------
|"A"  |"B"  |"C"  |"D"  |
-------------------------
|1    |2    |3    |4    |
-------------------------

    Crie um DataFrame e especifique um esquema:

    # Create a DataFrame and specify a schema
from snowflake.snowpark.types import IntegerType, StringType, StructType, StructField
schema = StructType([StructField("a", IntegerType()), StructField("b", StringType())])
df4 = session.create_dataframe([[1, "snow"], [3, "flake"]], schema)
df4.show()
# To return the DataFrame as a table in a Python worksheet use return instead of show()
# return df4
    Copy
    ---------------
|"A"  |"B"    |
---------------
|1    |snow   |
|3    |flake  |
---------------

  • Para criar um DataFrame contendo um intervalo de valores, chame o método range:

    # Create a DataFrame from a range
# The DataFrame contains rows with values 1, 3, 5, 7, and 9 respectively.
df_range = session.range(1, 10, 2).to_df("a")
df_range.show()
# To return the DataFrame as a table in a Python worksheet use return instead of show()
# return df_range
    Copy
    -------
|"A"  |
-------
|1    |
|3    |
|5    |
|7    |
|9    |
-------

  • Para criar um DataFrame para armazenar os dados de um arquivo em um estágio, use a propriedade read para obter um objeto DataFrameReader. No objeto DataFrameReader, chame o método correspondente ao formato dos dados no arquivo:

    from snowflake.snowpark.types import StructType, StructField, StringType, IntegerType

# Create DataFrames from data in a stage.
df_json = session.read.json("@my_stage2/data1.json")
df_catalog = session.read.schema(StructType([StructField("name", StringType()), StructField("age", IntegerType())])).csv("@stage/some_dir")
    Copy

  • Para criar um DataFrame para armazenar os resultados de uma consulta SQL, chame o método sql:

    # Create a DataFrame from a SQL query
df_sql = session.sql("SELECT name from sample_product_data")
df_sql.show()
# To return the DataFrame as a table in a Python worksheet use return instead of show()
# return df_sql
    Copy
    --------------
|"NAME"      |
--------------
|Product 1   |
|Product 1A  |
|Product 1B  |
|Product 2   |
|Product 2A  |
|Product 2B  |
|Product 3   |
|Product 3A  |
|Product 3B  |
|Product 4   |
--------------

É possível usar o método sql para executar instruções SELECT que recuperam dados de tabelas e arquivos preparados, mas o uso do método table e a propriedade read oferece melhor destaque de sintaxe, destaque de erros e preenchimento inteligente de código em ferramentas de desenvolvimento.

Especificação de como o conjunto de dados deve ser transformado

Para especificar quais colunas selecionar e como filtrar, ordenar, agrupar etc. os resultados, chame os métodos DataFrame que transformam o conjunto de dados. Para identificar colunas nestes métodos, use a função col ou uma expressão que seja avaliada como coluna. Consulte Como especificar colunas e expressões.

Por exemplo:

  • Para especificar quais linhas devem ser retornadas, chame o método filter:

    # Import the col function from the functions module.
# Python worksheets import this function by default
from snowflake.snowpark.functions import col

# Create a DataFrame for the rows with the ID 1
# in the "sample_product_data" table.

# This example uses the == operator of the Column object to perform an
# equality check.
df = session.table("sample_product_data").filter(col("id") == 1)
df.show()
# To return the DataFrame as a table in a Python worksheet use return instead of show()
return df
    Copy
    ------------------------------------------------------------------------------------
|"ID"  |"PARENT_ID"  |"CATEGORY_ID"  |"NAME"     |"SERIAL_NUMBER"  |"KEY"  |"3rd"  |
------------------------------------------------------------------------------------
|1     |0            |5              |Product 1  |prod-1           |1      |10     |
------------------------------------------------------------------------------------

  • Para especificar as colunas que devem ser selecionadas, chame o método select:

    # Import the col function from the functions module.
from snowflake.snowpark.functions import col

# Create a DataFrame that contains the id, name, and serial_number
# columns in the "sample_product_data" table.
df = session.table("sample_product_data").select(col("id"), col("name"), col("serial_number"))
df.show()
# To return the DataFrame as a table in a Python worksheet use return instead of show()
return df
    Copy
    ---------------------------------------
|"ID"  |"NAME"      |"SERIAL_NUMBER"  |
---------------------------------------
|1     |Product 1   |prod-1           |
|2     |Product 1A  |prod-1-A         |
|3     |Product 1B  |prod-1-B         |
|4     |Product 2   |prod-2           |
|5     |Product 2A  |prod-2-A         |
|6     |Product 2B  |prod-2-B         |
|7     |Product 3   |prod-3           |
|8     |Product 3A  |prod-3-A         |
|9     |Product 3B  |prod-3-B         |
|10    |Product 4   |prod-4           |
---------------------------------------

  • Você também pode consultar colunas como esta:

    # Import the col function from the functions module.
from snowflake.snowpark.functions import col

df_product_info = session.table("sample_product_data")
df1 = df_product_info.select(df_product_info["id"], df_product_info["name"], df_product_info["serial_number"])
df2 = df_product_info.select(df_product_info.id, df_product_info.name, df_product_info.serial_number)
df3 = df_product_info.select("id", "name", "serial_number")
    Copy

Cada método retorna um novo objeto DataFrame que foi transformado. O método não afeta o objeto DataFrame original. Se você quiser aplicar múltiplas transformações, você pode encadear chamadas a métodos, chamando cada método de transformação subsequente sobre o novo objeto DataFrame retornado pela chamada ao método anterior.

Estes métodos de transformação especificam como criar a instrução SQL e não recuperam dados do banco de dados Snowflake. Os métodos de ação descritos em Como executar uma ação para avaliar um DataFrame realizam a recuperação de dados.

Junção de DataFrames

Para juntar objetos DataFrame, chame o método join:

# Create two DataFrames to join
df_lhs = session.create_dataframe([["a", 1], ["b", 2]], schema=["key", "value1"])
df_rhs = session.create_dataframe([["a", 3], ["b", 4]], schema=["key", "value2"])
# Create a DataFrame that joins the two DataFrames
# on the column named "key".
df_lhs.join(df_rhs, df_lhs.col("key") == df_rhs.col("key")).select(df_lhs["key"].as_("key"), "value1", "value2").show()
# To return the DataFrame as a table in a Python worksheet use return instead of show()
return df_lhs.join(df_rhs, df_lhs.col("key") == df_rhs.col("key")).select(df_lhs["key"].as_("key"), "value1", "value2")
Copy
-------------------------------
|"KEY"  |"VALUE1"  |"VALUE2"  |
-------------------------------
|a      |1         |3         |
|b      |2         |4         |
-------------------------------

Se ambos DataFrames tiverem a mesma coluna para junção, você pode usar a seguinte sintaxe de exemplo:

# Create two DataFrames to join
df_lhs = session.create_dataframe([["a", 1], ["b", 2]], schema=["key", "value1"])
df_rhs = session.create_dataframe([["a", 3], ["b", 4]], schema=["key", "value2"])
# If both dataframes have the same column "key", the following is more convenient.
df_lhs.join(df_rhs, ["key"]).show()
# To return the DataFrame as a table in a Python worksheet use return instead of show()
return df_lhs.join(df_rhs, ["key"])
Copy
-------------------------------
|"KEY"  |"VALUE1"  |"VALUE2"  |
-------------------------------
|a      |1         |3         |
|b      |2         |4         |
-------------------------------

Você também pode usar o operador & para conectar expressões de junção:

# Create two DataFrames to join
df_lhs = session.create_dataframe([["a", 1], ["b", 2]], schema=["key", "value1"])
df_rhs = session.create_dataframe([["a", 3], ["b", 4]], schema=["key", "value2"])
# Use & operator connect join expression. '|' and ~ are similar.
df_joined_multi_column = df_lhs.join(df_rhs, (df_lhs.col("key") == df_rhs.col("key")) & (df_lhs.col("value1") < df_rhs.col("value2"))).select(df_lhs["key"].as_("key"), "value1", "value2")
df_joined_multi_column.show()
# To return the DataFrame as a table in a Python worksheet use return instead of show()
return df_joined_multi_column
Copy
-------------------------------
|"KEY"  |"VALUE1"  |"VALUE2"  |
-------------------------------
|a      |1         |3         |
|b      |2         |4         |
-------------------------------

Se você quiser realizar uma autojunção, você deverá copiar o DataFrame:

# copy the DataFrame if you want to do a self-join
from copy import copy

# Create two DataFrames to join
df_lhs = session.create_dataframe([["a", 1], ["b", 2]], schema=["key", "value1"])
df_rhs = session.create_dataframe([["a", 3], ["b", 4]], schema=["key", "value2"])
df_lhs_copied = copy(df_lhs)
df_self_joined = df_lhs.join(df_lhs_copied, (df_lhs.col("key") == df_lhs_copied.col("key")) & (df_lhs.col("value1") == df_lhs_copied.col("value1")))
Copy

Quando há colunas sobrepostas nos DataFrames, o Snowpark irá anexar um prefixo gerado aleatoriamente para as colunas no resultado da junção:

# Create two DataFrames to join
df_lhs = session.create_dataframe([["a", 1], ["b", 2]], schema=["key", "value1"])
df_rhs = session.create_dataframe([["a", 3], ["b", 4]], schema=["key", "value2"])
df_lhs.join(df_rhs, df_lhs.col("key") == df_rhs.col("key")).show()
# To return the DataFrame as a table in a Python worksheet use return instead of show()
return df_lhs.join(df_rhs, df_lhs.col("key") == df_rhs.col("key"))
Copy
-----------------------------------------------------
|"l_av5t_KEY"  |"VALUE1"  |"r_1p6k_KEY"  |"VALUE2"  |
-----------------------------------------------------
|a             |1         |a             |3         |
|b             |2         |b             |4         |
-----------------------------------------------------

Você pode renomear as colunas sobrepostas usando Column.alias:

# Create two DataFrames to join
df_lhs = session.create_dataframe([["a", 1], ["b", 2]], schema=["key", "value1"])
df_rhs = session.create_dataframe([["a", 3], ["b", 4]], schema=["key", "value2"])
df_lhs.join(df_rhs, df_lhs.col("key") == df_rhs.col("key")).select(df_lhs["key"].alias("key1"), df_rhs["key"].alias("key2"), "value1", "value2").show()
# To return the DataFrame as a table in a Python worksheet use return instead of show()
return df_lhs.join(df_rhs, df_lhs.col("key") == df_rhs.col("key")).select(df_lhs["key"].alias("key1"), df_rhs["key"].alias("key2"), "value1", "value2")
Copy
-----------------------------------------
|"KEY1"  |"KEY2"  |"VALUE1"  |"VALUE2"  |
-----------------------------------------
|a       |a       |1         |3         |
|b       |b       |2         |4         |
-----------------------------------------

Para evitar prefixos aleatórios, você também pode especificar um sufixo para anexar às colunas sobrepostas:

# Create two DataFrames to join
df_lhs = session.create_dataframe([["a", 1], ["b", 2]], schema=["key", "value1"])
df_rhs = session.create_dataframe([["a", 3], ["b", 4]], schema=["key", "value2"])
df_lhs.join(df_rhs, df_lhs.col("key") == df_rhs.col("key"), lsuffix="_left", rsuffix="_right").show()
# To return the DataFrame as a table in a Python worksheet use return instead of show()
return df_lhs.join(df_rhs, df_lhs.col("key") == df_rhs.col("key"), lsuffix="_left", rsuffix="_right")
Copy
--------------------------------------------------
|"KEY_LEFT"  |"VALUE1"  |"KEY_RIGHT"  |"VALUE2"  |
--------------------------------------------------
|a           |1         |a            |3         |
|b           |2         |b            |4         |
--------------------------------------------------

Estes exemplos usam DataFrame.col para especificar as colunas a serem usadas na junção. Consulte Como especificar colunas e expressões para obter mais maneiras de especificar colunas.

Se você precisar unir uma tabela consigo mesma em diferentes colunas, você não pode realizar a autojunção com um único DataFrame. Os exemplos a seguir que usam um único DataFrame para realizar uma autojunção, que falha porque as expressões da coluna para "id" estão presentes nos lados esquerdo e direito da junção:

from snowflake.snowpark.exceptions import SnowparkJoinException

df = session.table("sample_product_data")
# This fails because columns named "id" and "parent_id"
# are in the left and right DataFrames in the join.
try:
  df_joined = df.join(df, col("id") == col("parent_id")) # fails
except SnowparkJoinException as e:
  print(e.message)
Copy
You cannot join a DataFrame with itself because the column references cannot be resolved correctly. Instead, create a copy of the DataFrame with copy.copy(), and join the DataFrame with this copy.

# This fails because columns named "id" and "parent_id"
# are in the left and right DataFrames in the join.
try:
  df_joined = df.join(df, df["id"] == df["parent_id"])   # fails
except SnowparkJoinException as e:
  print(e.message)
Copy
You cannot join a DataFrame with itself because the column references cannot be resolved correctly. Instead, create a copy of the DataFrame with copy.copy(), and join the DataFrame with this copy.

Em vez disso, use o método clone copy() incluído com o Python para criar um clone do objeto DataFrame, e use os dois objetos DataFrame para realizar a junção:

from copy import copy

# Create a DataFrame object for the "sample_product_data" table for the left-hand side of the join.
df_lhs = session.table("sample_product_data")
# Clone the DataFrame object to use as the right-hand side of the join.
df_rhs = copy(df_lhs)

# Create a DataFrame that joins the two DataFrames
# for the "sample_product_data" table on the
# "id" and "parent_id" columns.
df_joined = df_lhs.join(df_rhs, df_lhs.col("id") == df_rhs.col("parent_id"))
df_joined.count()
Copy

Como especificar colunas e expressões

Ao chamar esses métodos de transformação, pode ser necessário especificar colunas ou expressões que utilizam colunas. Por exemplo, ao chamar o método select, você precisa especificar as colunas a serem selecionadas.

Para se referir a uma coluna, crie um objeto Column chamando a função col no módulo snowflake.snowpark.functions.

# Import the col function from the functions module.
from snowflake.snowpark.functions import col

df_product_info = session.table("sample_product_data").select(col("id"), col("name"))
df_product_info.show()
# To return the DataFrame as a table in a Python worksheet use return instead of show()
return df_product_info
Copy
---------------------
|"ID"  |"NAME"      |
---------------------
|1     |Product 1   |
|2     |Product 1A  |
|3     |Product 1B  |
|4     |Product 2   |
|5     |Product 2A  |
|6     |Product 2B  |
|7     |Product 3   |
|8     |Product 3A  |
|9     |Product 3B  |
|10    |Product 4   |
---------------------

Nota

Para criar um objeto Column para um literal, consulte Como usar literais como objetos de coluna.

Ao especificar um filtro, projeção, condição de junção, etc., você pode usar objetos Column em uma expressão. Por exemplo:

  • Você pode usar objetos Column com o método filter para especificar uma condição de filtro:

    # Specify the equivalent of "WHERE id = 20"
# in a SQL SELECT statement.
df_filtered = df.filter(col("id") == 20)
    Copy
    df = session.create_dataframe([[1, 3], [2, 10]], schema=["a", "b"])
# Specify the equivalent of "WHERE a + b < 10"
# in a SQL SELECT statement.
df_filtered = df.filter((col("a") + col("b")) < 10)
df_filtered.show()
# To return the DataFrame as a table in a Python worksheet use return instead of show()
return df_filtered
    Copy
    -------------
|"A"  |"B"  |
-------------
|1    |3    |
-------------

  • Você pode usar objetos Column com o método select para definir um alias:

    df = session.create_dataframe([[1, 3], [2, 10]], schema=["a", "b"])
# Specify the equivalent of "SELECT b * 10 AS c"
# in a SQL SELECT statement.
df_selected = df.select((col("b") * 10).as_("c"))
df_selected.show()
# To return the DataFrame as a table in a Python worksheet use return instead of show()
return df_selected
    Copy
    -------
|"C"  |
-------
|30   |
|100  |
-------

  • Você pode usar objetos Column com o método join para definir uma condição de junção:

    dfX = session.create_dataframe([[1], [2]], schema=["a_in_X"])
dfY = session.create_dataframe([[1], [3]], schema=["b_in_Y"])
# Specify the equivalent of "X JOIN Y on X.a_in_X = Y.b_in_Y"
# in a SQL SELECT statement.
df_joined = dfX.join(dfY, col("a_in_X") == col("b_in_Y")).select(dfX["a_in_X"].alias("the_joined_column"))
df_joined.show()
# To return the DataFrame as a table in a Python worksheet use return instead of show()
return df_joined
    Copy
    -----------------------
|"THE_JOINED_COLUMN"  |
-----------------------
|1                    |
-----------------------

Ao se referir a colunas em dois objetos DataFrame diferentes que têm o mesmo nome (por exemplo, ao unir os DataFrames naquela coluna), você pode usar o método DataFrame.col em um objeto DataFrame para se referir a uma coluna naquele objeto (por exemplo, df1.col("name") e df2.col("name")).

O exemplo a seguir demonstra como usar o método DataFrame.col para se referir a uma coluna em um DataFrame específico. O exemplo une dois objetos DataFrame, ambos contendo uma coluna chamada key. O exemplo usa o método Column.as para mudar os nomes das colunas no DataFrame recém-criado.

# Create two DataFrames to join
df_lhs = session.create_dataframe([["a", 1], ["b", 2]], schema=["key", "value"])
df_rhs = session.create_dataframe([["a", 3], ["b", 4]], schema=["key", "value"])
# Create a DataFrame that joins two other DataFrames (df_lhs and df_rhs).
# Use the DataFrame.col method to refer to the columns used in the join.
df_joined = df_lhs.join(df_rhs, df_lhs.col("key") == df_rhs.col("key")).select(df_lhs.col("key").as_("key"), df_lhs.col("value").as_("L"), df_rhs.col("value").as_("R"))
df_joined.show()
# To return the DataFrame as a table in a Python worksheet use return instead of show()
return df_joined
Copy
---------------------
|"KEY"  |"L"  |"R"  |
---------------------
|a      |1    |3    |
|b      |2    |4    |
---------------------

Como utilizar aspas duplas em torno de identificadores de objetos (nomes de tabela, nomes de coluna, etc.)

Os nomes de bancos de dados, esquemas, tabelas e estágios que você especificar devem estar em conformidade com requisitos de identificadores do Snowflake.

Crie uma tabela com colunas que diferenciam maiúsculas de minúsculas:

session.sql("""
  create or replace temp table "10tablename"(
  id123 varchar, -- case insensitive because it's not quoted.
  "3rdID" varchar, -- case sensitive.
  "id with space" varchar -- case sensitive.
)""").collect()
# Add return to the statement to return the collect() results in a Python worksheet
Copy
[Row(status='Table 10tablename successfully created.')]

Em seguida, acrescente valores à tabela:

session.sql("""insert into "10tablename" (id123, "3rdID", "id with space") values ('a', 'b', 'c')""").collect()
# Add return to the statement to return the collect() results in a Python worksheet
Copy
[Row(number of rows inserted=1)]

Depois crie um DataFrame para a tabela e consulte a tabela:

df = session.table('"10tablename"')
df.show()
# To return the DataFrame as a table in a Python worksheet use return instead of show()
return df
Copy
---------------------------------------
|"ID123"  |"3rdID"  |"id with space"  |
---------------------------------------
|a        |b        |c                |
---------------------------------------

Quando você especifica um nome, o Snowflake considera que o nome está em maiúsculas. Por exemplo, as chamadas a seguir são equivalentes:

df.select(col("id123")).collect()
# Prepend a return statement to return the collect() results in a Python worksheet
Copy
[Row(ID123='a')]

Se o nome não estiver de acordo com os requisitos de identificador, deve-se usar aspas duplas (") em torno do nome. Use uma barra invertida (\) para escapar o caractere de aspas duplas dentro de uma cadeia de caracteres literal. Por exemplo, como o nome da tabela a seguir não começa com uma letra ou um sublinhado, você deve usar aspas duplas em torno do nome:

df = session.table("\"10tablename\"")
Copy

Alternativamente, você pode usar aspas simples em vez de barras invertidas para escapar o caractere de aspas duplas dentro de uma cadeia de caracteres literal.

df = session.table('"10tablename"')
Copy

Observe que ao especificar o nome de uma coluna, você não precisa usar aspas duplas em torno do nome. A biblioteca do Snowpark coloca automaticamente o nome da coluna entre aspas duplas caso o nome não cumpra com os requisitos de identificador:

df.select(col("3rdID")).collect()
# Prepend a return statement to return the collect() results in a Python worksheet
Copy
[Row(3rdID='b')]

Como outro exemplo, as chamadas a seguir são equivalentes:

df.select(col("id with space")).collect()
# Prepend a return statement to return the collect() results in a Python worksheet
Copy
[Row(id with space='c')]

df.select(col("\"id with space\"")).collect()
# Prepend a return statement to return the collect() results in a Python worksheet
Copy
[Row(id with space='c')]

Se você já tiver acrescentado aspas duplas em torno de um nome de coluna, a biblioteca não inserirá aspas duplas adicionais em torno do nome.

Em alguns casos, o nome da coluna pode conter caracteres de aspas duplas:

session.sql('''
  create or replace temp table quoted(
  "name_with_""air""_quotes" varchar,
  """column_name_quoted""" varchar
)''').collect()
# Prepend a return statement to return the collect() results in a Python worksheet
Copy
[Row(status='Table QUOTED successfully created.')]

session.sql('''insert into quoted ("name_with_""air""_quotes", """column_name_quoted""") values ('a', 'b')''').collect()
# Prepend a return statement to return the collect() results in a Python worksheet
Copy
[Row(number of rows inserted=1)]

Como explicado em Requisitos para identificadores, para cada caractere de aspas duplas dentro de um identificador de aspas duplas, deve-se usar dois caracteres de aspas duplas (por exemplo, "name_with_""air""_quotes" e """column_name_quoted"""):

df_table = session.table("quoted")
df_table.select("\"name_with_\"\"air\"\"_quotes\"").collect()
# Prepend a return statement to return the collect() results in a Python worksheet
Copy
[Row(name_with_"air"_quotes='a')]

df_table.select("\"\"\"column_name_quoted\"\"\"").collect()
# Prepend a return statement to return the collect() results in a Python worksheet
Copy
[Row("column_name_quoted"='b')]

Quando um identificador é incluído entre aspas duplas (se você adicionou explicitamente as aspas ou se a biblioteca adicionou as aspas para você), o Snowflake trata o identificador diferenciando maiúsculas de minúsculas:

# The following calls are NOT equivalent!
# The Snowpark library adds double quotes around the column name,
# which makes Snowflake treat the column name as case-sensitive.
df.select(col("id with space")).collect()
# Prepend a return statement to return the collect() results in a Python worksheet
Copy
[Row(id with space='c')]

Em comparação com este exemplo:

from snowflake.snowpark.exceptions import SnowparkSQLException
try:
  df.select(col("ID WITH SPACE")).collect()
except SnowparkSQLException as e:
  print(e.message)
Copy
000904 (42000): SQL compilation error: error line 1 at position 7
invalid identifier '"ID WITH SPACE"'

Como usar literais como objetos de coluna

Para usar um literal em um método que toma um objeto Column como argumento, crie um objeto Column para o literal passando o literal para a função lit no módulo snowflake.snowpark.functions. Por exemplo:

# Import for the lit and col functions.
from snowflake.snowpark.functions import col, lit

# Show the first 10 rows in which num_items is greater than 5.
# Use `lit(5)` to create a Column object for the literal 5.
df_filtered = df.filter(col("num_items") > lit(5))
Copy

Como converter um objeto de coluna para um tipo específico

Para converter um objeto Column a um tipo específico, chame o método cast e passe um objeto do módulo snowflake.snowpark.types. Por exemplo, para converter um literal para um NUMBER com uma precisão de 5 e escala de 2:

# Import for the lit function.
from snowflake.snowpark.functions import lit

 # Import for the DecimalType class.
from snowflake.snowpark.types import DecimalType

decimal_value = lit(0.05).cast(DecimalType(5,2))
Copy

Como encadear chamadas a métodos

Como cada método que transforma um objeto DataFrame returna um novo objeto DataFrame que tem a transformação aplicada, você pode encadear chamadas a métodos para produzir um novo DataFrame que é transformado de maneiras adicionais.

O exemplo a seguir retorna um DataFrame que está configurado para:

  • Consultar a tabela sample_product_data.

  • Retornar a linha com id = 1.

  • Selecionar as colunas name e serial_number.

    df_product_info = session.table("sample_product_data").filter(col("id") == 1).select(col("name"), col("serial_number"))
df_product_info.show()
# To return the DataFrame as a table in a Python worksheet use return instead of show()
return df_product_info
    Copy
    -------------------------------
|"NAME"     |"SERIAL_NUMBER"  |
-------------------------------
|Product 1  |prod-1           |
-------------------------------

Neste exemplo:

  • session.table("sample_product_data") retorna um DataFrame para a tabela sample_product_data.

    Embora o DataFrame ainda não contenha os dados da tabela, o objeto contém as definições das colunas da tabela.

  • filter(col("id") == 1) retorna um DataFrame para a tabela sample_product_data que está preparada para retornar a linha com id = 1.

    Observe que o DataFrame ainda não contém a linha correspondente da tabela. A linha correspondente não é obtida até que você chame um método de ação.

  • select(col("name"), col("serial_number")) retorna um DataFrame que contém as colunas name e serial_number para a linha da tabela sample_product_data que tem id = 1.

A ordem das chamadas é importante quando você encadeia chamadas de métodos. Cada chamada de método retorna um DataFrame que foi transformado. Certifique-se de que as chamadas subsequentes funcionem com os DataFrame transformados.

Ao usar o Snowpark Python, talvez seja necessário fazer as chamadas ao método select e filter em uma ordem diferente daquela em que você usaria as palavras-chave equivalentes (SELECT e WHERE) em uma instrução SQL.

Como obter definições das colunas

Para obter a definição das colunas no conjunto de dados para o DataFrame, chame a propriedade schema. Esse método retorna um objeto StructType que contém uma list de StructField objetos. Cada objeto StructField contém a definição de uma coluna.

# Import the StructType
from snowflake.snowpark.types import *
# Get the StructType object that describes the columns in the
# underlying rowset.
table_schema = session.table("sample_product_data").schema
table_schema
StructType([StructField('ID', LongType(), nullable=True), StructField('PARENT_ID', LongType(), nullable=True), StructField('CATEGORY_ID', LongType(), nullable=True), StructField('NAME', StringType(), nullable=True), StructField('SERIAL_NUMBER', StringType(), nullable=True), StructField('KEY', LongType(), nullable=True), StructField('"3rd"', LongType(), nullable=True)])
Copy

No objeto devolvido StructType, os nomes das colunas são sempre normalizados. Identificadores sem aspas são retornados em maiúsculas, e identificadores entre aspas são devolvidos na caixa em que foram definidos.

O exemplo a seguir cria um DataFrame contendo as colunas ID e 3rd. Para o nome de coluna 3rd, a biblioteca do Snowpark coloca o nome automaticamente entre aspas duplas ("3rd") porque o nome não cumpre os requisitos para um identificador.

O exemplo chama a propriedade schema e depois chama a propriedade names no objeto StructType retornado para obter uma list de nomes de coluna. Os nomes são normalizados no StructType retornado pela propriedade schema.

# Create a DataFrame containing the "id" and "3rd" columns.
df_selected_columns = session.table("sample_product_data").select(col("id"), col("3rd"))
# Print out the names of the columns in the schema.
# This prints List["ID", "\"3rd\""]
df_selected_columns.schema.names
Copy
['ID', '"3rd"']

Como executar uma ação para avaliar um DataFrame

Como mencionado anteriormente, o DataFrame é avaliado lentamente, o que significa que a instrução SQL não é enviada ao servidor para execução até que você execute uma ação. Uma ação faz com que o DataFrame seja avaliado e envia a instrução SQL correspondente para o servidor para execução.

Os seguintes métodos executam uma ação:

Classe

Método

Descrição

DataFrame

collect

Avalia o DataFrame e retorna o conjunto de dados resultante como uma list de objetos Row.

DataFrame

count

Avalia o DataFrame e retorna o número de linhas.

DataFrame

show

Avalia o DataFrame e imprime as linhas no console. Este método limita o número de linhas a 10 (por padrão).

DataFrameWriter

save_as_table

Armazena os dados no DataFrame para a tabela especificada. Consulte Como salvar dados em uma tabela.

Por exemplo, para executar uma consulta em uma tabela e retornar os resultados, chame o método collect:

# Create a DataFrame with the "id" and "name" columns from the "sample_product_data" table.
# This does not execute the query.
df = session.table("sample_product_data").select(col("id"), col("name"))

# Send the query to the server for execution and
# return a list of Rows containing the results.
results = df.collect()
# Use a return statement to return the collect() results in a Python worksheet
# return results
Copy

Para executar a consulta e retornar o número de resultados, chame o método count:

# Create a DataFrame for the "sample_product_data" table.
df_products = session.table("sample_product_data")

# Send the query to the server for execution and
# print the count of rows in the table.
print(df_products.count())
12
Copy

Para executar uma consulta e imprimir os resultados para o console, chame o método show:

# Create a DataFrame for the "sample_product_data" table.
df_products = session.table("sample_product_data")

# Send the query to the server for execution and
# print the results to the console.
# The query limits the number of rows to 10 by default.
df_products.show()
# To return the DataFrame as a table in a Python worksheet use return instead of show()
return df_products
Copy
-------------------------------------------------------------------------------------
|"ID"  |"PARENT_ID"  |"CATEGORY_ID"  |"NAME"      |"SERIAL_NUMBER"  |"KEY"  |"3rd"  |
-------------------------------------------------------------------------------------
|1     |0            |5              |Product 1   |prod-1           |1      |10     |
|2     |1            |5              |Product 1A  |prod-1-A         |1      |20     |
|3     |1            |5              |Product 1B  |prod-1-B         |1      |30     |
|4     |0            |10             |Product 2   |prod-2           |2      |40     |
|5     |4            |10             |Product 2A  |prod-2-A         |2      |50     |
|6     |4            |10             |Product 2B  |prod-2-B         |2      |60     |
|7     |0            |20             |Product 3   |prod-3           |3      |70     |
|8     |7            |20             |Product 3A  |prod-3-A         |3      |80     |
|9     |7            |20             |Product 3B  |prod-3-B         |3      |90     |
|10    |0            |50             |Product 4   |prod-4           |4      |100    |
-------------------------------------------------------------------------------------

Para limitar o número de linhas a 20:

# Create a DataFrame for the "sample_product_data" table.
df_products = session.table("sample_product_data")

# Limit the number of rows to 20, rather than 10.
df_products.show(20)
# All rows are returned when you use return in a Python worksheet to return the DataFrame as a table
return df_products
Copy
-------------------------------------------------------------------------------------
|"ID"  |"PARENT_ID"  |"CATEGORY_ID"  |"NAME"      |"SERIAL_NUMBER"  |"KEY"  |"3rd"  |
-------------------------------------------------------------------------------------
|1     |0            |5              |Product 1   |prod-1           |1      |10     |
|2     |1            |5              |Product 1A  |prod-1-A         |1      |20     |
|3     |1            |5              |Product 1B  |prod-1-B         |1      |30     |
|4     |0            |10             |Product 2   |prod-2           |2      |40     |
|5     |4            |10             |Product 2A  |prod-2-A         |2      |50     |
|6     |4            |10             |Product 2B  |prod-2-B         |2      |60     |
|7     |0            |20             |Product 3   |prod-3           |3      |70     |
|8     |7            |20             |Product 3A  |prod-3-A         |3      |80     |
|9     |7            |20             |Product 3B  |prod-3-B         |3      |90     |
|10    |0            |50             |Product 4   |prod-4           |4      |100    |
|11    |10           |50             |Product 4A  |prod-4-A         |4      |100    |
|12    |10           |50             |Product 4B  |prod-4-B         |4      |100    |
-------------------------------------------------------------------------------------

Nota

Se você chamar a propriedade schema para obter as definições das colunas no DataFrame, você não precisará chamar um método de ação.

Como salvar dados em uma tabela

Para salvar o conteúdo de um DataFrame em uma tabela:

  1. Chame a propriedade write para obter um objeto DataFrameWriter.

  2. Chame o método mode no objeto DataFrameWriter e especifique o modo. Para obter mais informações, consulte a documentação da API em. Esse método retorna um novo objeto DataFrameWriter que é configurado com o modo especificado.

  3. Chame o método save_as_table no objeto DataFrameWriter para salvar o conteúdo do DataFrame em uma tabela especificada.

Observe que não é necessário chamar um método separado (por exemplo, collect) para executar a instrução SQL que salva os dados na tabela.

Por exemplo:

df.write.mode("overwrite").save_as_table("table1")
Copy

Como criar uma exibição a partir de um DataFrame

Para criar uma exibição a partir de um DataFrame, chame o método create_or_replace_view, que imediatamente cria a nova exibição:

import os
database = os.environ["snowflake_database"]  # use your own database and schema
schema = os.environ["snowflake_schema"]
view_name = "my_view"
df.create_or_replace_view(f"{database}.{schema}.{view_name}")
Copy
[Row(status='View MY_VIEW successfully created.')]

Em uma planilha Python, porque você executa a planilha no contexto de um banco de dados e esquema, você pode executar o seguinte para criar uma exibição:

# Define a DataFrame
df_products = session.table("sample_product_data")
# Define a View name
view_name = "my_view"
# Create the view
df_products.create_or_replace_view(f"{view_name}")
# return the view name
return view_name + " successfully created"
my_view successfully created
Copy

As exibições que você cria ao chamar create_or_replace_view são persistentes. Se você não precisar mais dessa exibição, você pode excluir a exibição manualmente.

Alternativamente, use o método create_or_replace_temp_view, que cria uma exibição temporária. A exibição temporária só fica disponível na sessão em que ela é criada.

Como trabalhar com arquivos em um estágio

Esta seção explica como consultar dados em um arquivo em um estágio do Snowflake. Para outras operações em arquivos, use instruções SQL.

Para consultar dados em arquivos em um estágio do Snowflake, use a classe DataFrameReader:

  1. Chame o método read na classe Session para acessar um objeto DataFrameReader.

  2. Se os arquivos estiverem no formato CSV, descreva os campos no arquivo. Para fazer isso:

    1. Crie um objeto StructType que consiste em uma list de objetos StructField descrevendo os campos no arquivo.

    2. Para cada objeto StructField, especifique o seguinte:

      • O nome do campo.

      • O tipo de dados do campo (especificado como um objeto no módulo snowflake.snowpark.types).

      • Se o campo é ou não anulável.

      Por exemplo:

      from snowflake.snowpark.types import *

schema_for_data_file = StructType([
                          StructField("id", StringType()),
                          StructField("name", StringType())
                       ])
      Copy

    3. Chame a propriedade schema no objeto DataFrameReader, passando o objeto StructType.

      Por exemplo:

      df_reader = session.read.schema(schema_for_data_file)
      Copy

      A propriedade schema retorna um objeto DataFrameReader que é configurado para ler arquivos contendo os campos especificados.

      Note que você não precisa fazer isso para arquivos em outros formatos (como JSON). Para esses arquivos, o DataFrameReader trata os dados como um único campo do tipo VARIANT com o nome de campo $1.

  3. Se você precisar especificar informações adicionais sobre como os dados devem ser lidos (por exemplo, que os dados estão comprimidos ou que um arquivo CSV usa um ponto-e-vírgula em vez de uma vírgula para delimitar campos), chame os métodos option ou options do objeto DataFrameReader.

    O método option obtém um nome e um valor da opção que você deseja definir e permite combinar múltiplas chamadas encadeadas, enquanto o método options obtém um dicionário dos nomes das opções e seus valores correspondentes.

    Para os nomes e valores das opções de formato do arquivo, veja a documentação documentação sobre CREATE FILE FORMAT.

    Você também pode definir as opções de cópia descritas na documentação COPY INTO TABLE. Observe que a definição de opções de cópia pode resultar em uma estratégia de execução mais cara quando você obtém os dados para o DataFrame.

    O exemplo a seguir configura o objeto DataFrameReader para consultar dados em um arquivo CSV que não é compactado e que usa um ponto-e-vírgula como delimitador de campo.

    df_reader = df_reader.option("field_delimiter", ";").option("COMPRESSION", "NONE")
    Copy

    Os métodos option e options retornam um objeto DataFrameReader que é configurado com as opções especificadas.

  4. Chame o método correspondente ao formato do arquivo (por exemplo, o método csv) passando o local do arquivo.

    df = df_reader.csv("@s3_ts_stage/emails/data_0_0_0.csv")
    Copy

    Os métodos correspondentes ao formato de um arquivo retornam um objeto DataFrame que é configurado para manter os dados nesse arquivo.

  5. Use os métodos do objeto DataFrame para realizar quaisquer transformações necessárias no conjunto de dados (por exemplo, selecionar campos específicos, filtrar linhas, etc.).

    Por exemplo, para extrair o elemento color de um arquivo JSON no estágio chamado my_stage:

    # Import the sql_expr function from the functions module.
from snowflake.snowpark.functions import sql_expr

df = session.read.json("@my_stage").select(sql_expr("$1:color"))
    Copy

    Como explicado anteriormente, para arquivos em formatos diferentes de CSV (por exemplo JSON), o DataFrameReader trata os dados no arquivo como uma única coluna VARIANT com o nome $1.

    Esse exemplo usa a função sql_expr no módulo snowflake.snowpark.functions para especificar o caminho para o elemento color.

    Observe que a função sql_expr não interpreta ou modifica o argumento de entrada. A função só permite formar expressões e trechos em SQL que ainda não são suportados pela API do Snowpark.

  6. Chame um método de ação para consultar os dados no arquivo.

    Como é o caso de DataFrames para tabelas, os dados não são obtidos no DataFrame até que você chame um método de ação.

Como trabalhar com dados semiestruturados

Usando um DataFrame, você pode consultar e acessar dados semiestruturados (por exemplo, dados de JSON). As próximas seções explicam como trabalhar com dados semiestruturados em um DataFrame.

Nota

Os exemplos nestas seções usam os dados de exemplo em Amostra de dados usados em exemplos.

Como percorrer dados semiestruturados

Para se referir a um campo ou elemento específico em dados semiestruturados, use os seguintes métodos do objeto Column:

  • Use o atributo col_object["<field_name>"] para retornar um objeto Column para o campo em um OBJECT (ou um VARIANT que contenha um OBJECT).

  • Use col_object[<index>] para retornar um objeto Column para um elemento em uma ARRAY (ou um VARIANT que contenha uma ARRAY).

Nota

Se o nome do campo ou elementos no caminho forem irregulares e dificultarem o uso da indexação descrita acima, você pode usar get, get_ignore_case ou get_path como alternativa.

Por exemplo, o seguinte código seleciona o campo dealership nos objetos na coluna src dos dados de exemplo:

from snowflake.snowpark.functions import col

df = session.table("car_sales")
df.select(col("src")["dealership"]).show()
Copy

O código imprime a seguinte saída:

----------------------------
|"""SRC""['DEALERSHIP']"   |
----------------------------
|"Valley View Auto Sales"  |
|"Tindel Toyota"           |
----------------------------

Nota

Os valores no DataFrame estão entre aspas duplas porque eles são retornados como literais de cadeias de caracteres. Para converter esses valores para um tipo específico, consulte Conversão explícita de valores em dados semiestruturados.

Você também pode encadear chamadas a métodos para percorrer um caminho para um campo ou elemento específico.

Por exemplo, o código a seguir seleciona o campo name no objeto salesperson:

df = session.table("car_sales")
df.select(df["src"]["salesperson"]["name"]).show()
Copy

O código imprime a seguinte saída:

------------------------------------
|"""SRC""['SALESPERSON']['NAME']"  |
------------------------------------
|"Frank Beasley"                   |
|"Greg Northrup"                   |
------------------------------------

Como outro exemplo, o código a seguir seleciona o primeiro elemento do campo vehicle, que contém uma array de veículos. O exemplo também seleciona o campo price do primeiro elemento.

df = session.table("car_sales")
df.select(df["src"]["vehicle"][0]).show()
df.select(df["src"]["vehicle"][0]["price"]).show()
Copy

O código imprime a seguinte saída:

---------------------------
|"""SRC""['VEHICLE'][0]"  |
---------------------------
|{                        |
|  "extras": [            |
|    "ext warranty",      |
|    "paint protection"   |
|  ],                     |
|  "make": "Honda",       |
|  "model": "Civic",      |
|  "price": "20275",      |
|  "year": "2017"         |
|}                        |
|{                        |
|  "extras": [            |
|    "ext warranty",      |
|    "rust proofing",     |
|    "fabric protection"  |
|  ],                     |
|  "make": "Toyota",      |
|  "model": "Camry",      |
|  "price": "23500",      |
|  "year": "2017"         |
|}                        |
---------------------------

------------------------------------
|"""SRC""['VEHICLE'][0]['PRICE']"  |
------------------------------------
|"20275"                           |
|"23500"                           |
------------------------------------

Como alternativa para acessar os campos da maneira mencionada acima, você pode usar as funções get, get_ignore_case, ou get_path se o nome do campo ou elementos no caminho forem irregulares.

Por exemplo, as duas linhas de código a seguir imprimem o valor de um campo especificado em um objeto:

from snowflake.snowpark.functions import get, get_path, lit

df.select(get(col("src"), lit("dealership"))).show()
df.select(col("src")["dealership"]).show()
Copy

Da mesma forma, as seguintes linhas de código imprimem o valor de um campo em um caminho especificado em um objeto:

df.select(get_path(col("src"), lit("vehicle[0].make"))).show()
df.select(col("src")["vehicle"][0]["make"]).show()
Copy

Conversão explícita de valores em dados semiestruturados

Por padrão, os valores de campos e elementos são retornados como literais de cadeias de caracteres (incluindo as aspas duplas), como mostrado nos exemplos acima.

Para evitar resultados inesperados, chame o método de converter para converter o valor para um tipo específico. Por exemplo, o seguinte código imprime os valores sem e com conversão:

# Import the objects for the data types, including StringType.
from snowflake.snowpark.types import *

df = session.table("car_sales")
df.select(col("src")["salesperson"]["id"]).show()
df.select(col("src")["salesperson"]["id"].cast(StringType())).show()
Copy

O código imprime a seguinte saída:

----------------------------------
|"""SRC""['SALESPERSON']['ID']"  |
----------------------------------
|"55"                            |
|"274"                           |
----------------------------------

---------------------------------------------------
|"CAST (""SRC""['SALESPERSON']['ID'] AS STRING)"  |
---------------------------------------------------
|55                                               |
|274                                              |
---------------------------------------------------

Como achatar uma array de objetos em linhas

Se você precisar “achatar” dados semiestruturados em um DataFrame (por exemplo, produzir uma linha para cada objeto na matriz), chame flatten usando o método join_table_function. Esse método é equivalente à função FLATTEN do SQL. Se você passar um caminho para um objeto ou array, o método retorna um DataFrame contendo uma linha para cada campo ou elemento no objeto ou array.

Por exemplo, nos dados de exemplo, src:customer é um conjunto de objetos que contém informações sobre um cliente. Cada objeto contém um campo name e address.

Se você passar esse caminho para a função flatten:

df = session.table("car_sales")
df.join_table_function("flatten", col("src")["customer"]).show()
Copy

o método retornará um DataFrame:

----------------------------------------------------------------------------------------------------------------------------------------------------------
|"SRC"                                      |"SEQ"  |"KEY"  |"PATH"  |"INDEX"  |"VALUE"                            |"THIS"                               |
----------------------------------------------------------------------------------------------------------------------------------------------------------
|{                                          |1      |NULL   |[0]     |0        |{                                  |[                                    |
|  "customer": [                            |       |       |        |         |  "address": "San Francisco, CA",  |  {                                  |
|    {                                      |       |       |        |         |  "name": "Joyce Ridgely",         |    "address": "San Francisco, CA",  |
|      "address": "San Francisco, CA",      |       |       |        |         |  "phone": "16504378889"           |    "name": "Joyce Ridgely",         |
|      "name": "Joyce Ridgely",             |       |       |        |         |}                                  |    "phone": "16504378889"           |
|      "phone": "16504378889"               |       |       |        |         |                                   |  }                                  |
|    }                                      |       |       |        |         |                                   |]                                    |
|  ],                                       |       |       |        |         |                                   |                                     |
|  "date": "2017-04-28",                    |       |       |        |         |                                   |                                     |
|  "dealership": "Valley View Auto Sales",  |       |       |        |         |                                   |                                     |
|  "salesperson": {                         |       |       |        |         |                                   |                                     |
|    "id": "55",                            |       |       |        |         |                                   |                                     |
|    "name": "Frank Beasley"                |       |       |        |         |                                   |                                     |
|  },                                       |       |       |        |         |                                   |                                     |
|  "vehicle": [                             |       |       |        |         |                                   |                                     |
|    {                                      |       |       |        |         |                                   |                                     |
|      "extras": [                          |       |       |        |         |                                   |                                     |
|        "ext warranty",                    |       |       |        |         |                                   |                                     |
|        "paint protection"                 |       |       |        |         |                                   |                                     |
|      ],                                   |       |       |        |         |                                   |                                     |
|      "make": "Honda",                     |       |       |        |         |                                   |                                     |
|      "model": "Civic",                    |       |       |        |         |                                   |                                     |
|      "price": "20275",                    |       |       |        |         |                                   |                                     |
|      "year": "2017"                       |       |       |        |         |                                   |                                     |
|    }                                      |       |       |        |         |                                   |                                     |
|  ]                                        |       |       |        |         |                                   |                                     |
|}                                          |       |       |        |         |                                   |                                     |
|{                                          |2      |NULL   |[0]     |0        |{                                  |[                                    |
|  "customer": [                            |       |       |        |         |  "address": "New York, NY",       |  {                                  |
|    {                                      |       |       |        |         |  "name": "Bradley Greenbloom",    |    "address": "New York, NY",       |
|      "address": "New York, NY",           |       |       |        |         |  "phone": "12127593751"           |    "name": "Bradley Greenbloom",    |
|      "name": "Bradley Greenbloom",        |       |       |        |         |}                                  |    "phone": "12127593751"           |
|      "phone": "12127593751"               |       |       |        |         |                                   |  }                                  |
|    }                                      |       |       |        |         |                                   |]                                    |
|  ],                                       |       |       |        |         |                                   |                                     |
|  "date": "2017-04-28",                    |       |       |        |         |                                   |                                     |
|  "dealership": "Tindel Toyota",           |       |       |        |         |                                   |                                     |
|  "salesperson": {                         |       |       |        |         |                                   |                                     |
|    "id": "274",                           |       |       |        |         |                                   |                                     |
|    "name": "Greg Northrup"                |       |       |        |         |                                   |                                     |
|  },                                       |       |       |        |         |                                   |                                     |
|  "vehicle": [                             |       |       |        |         |                                   |                                     |
|    {                                      |       |       |        |         |                                   |                                     |
|      "extras": [                          |       |       |        |         |                                   |                                     |
|        "ext warranty",                    |       |       |        |         |                                   |                                     |
|        "rust proofing",                   |       |       |        |         |                                   |                                     |
|        "fabric protection"                |       |       |        |         |                                   |                                     |
|      ],                                   |       |       |        |         |                                   |                                     |
|      "make": "Toyota",                    |       |       |        |         |                                   |                                     |
|      "model": "Camry",                    |       |       |        |         |                                   |                                     |
|      "price": "23500",                    |       |       |        |         |                                   |                                     |
|      "year": "2017"                       |       |       |        |         |                                   |                                     |
|    }                                      |       |       |        |         |                                   |                                     |
|  ]                                        |       |       |        |         |                                   |                                     |
|}                                          |       |       |        |         |                                   |                                     |
----------------------------------------------------------------------------------------------------------------------------------------------------------

A partir desse DataFrame, você poderá selecionar os campos name e address de cada objeto no campo VALUE:

df.join_table_function("flatten", col("src")["customer"]).select(col("value")["name"], col("value")["address"]).show()
Copy
-------------------------------------------------
|"""VALUE""['NAME']"   |"""VALUE""['ADDRESS']"  |
-------------------------------------------------
|"Joyce Ridgely"       |"San Francisco, CA"     |
|"Bradley Greenbloom"  |"New York, NY"          |
-------------------------------------------------

O código a seguir complementa o exemplo anterior ao converter os valores para um tipo específico e mudar os nomes das colunas:

df.join_table_function("flatten", col("src")["customer"]).select(col("value")["name"].cast(StringType()).as_("Customer Name"), col("value")["address"].cast(StringType()).as_("Customer Address")).show()
Copy
-------------------------------------------
|"Customer Name"     |"Customer Address"  |
-------------------------------------------
|Joyce Ridgely       |San Francisco, CA   |
|Bradley Greenbloom  |New York, NY        |
-------------------------------------------

Como executar instruções SQL

Para executar uma instrução SQL que você especificar, chame o método sql na classe Session e passe a instrução a ser executada. O método retorna um DataFrame.

Note que a instrução SQL não será executada até que você chame um método de ação.

# Get the list of the files in a stage.
# The collect() method causes this SQL statement to be executed.
session.sql("create or replace temp stage my_stage").collect()
Copy
# Prepend a return statement to return the collect() results in a Python worksheet
[Row(status='Stage area MY_STAGE successfully created.')]

stage_files_df = session.sql("ls @my_stage").collect()
# Prepend a return statement to return the collect() results in a Python worksheet
# Resume the operation of a warehouse.
# Note that you must call the collect method to execute
# the SQL statement.
session.sql("alter warehouse if exists my_warehouse resume if suspended").collect()
Copy
# Prepend a return statement to return the collect() results in a Python worksheet
[Row(status='Statement executed successfully.')]

# Set up a SQL statement to copy data from a stage to a table.
session.sql("copy into sample_product_data from @my_stage file_format=(type = csv)").collect()
# Prepend a return statement to return the collect() results in a Python worksheet
Copy
[Row(status='Copy executed with 0 files processed.')]

Se você quiser chamar métodos para transformar o DataFrame (por exemplo, filter, select, etc.), observe que esses métodos só funcionam se a instrução SQL subjacente for uma instrução SELECT. Os métodos de transformação não são suportados para outros tipos de instruções SQL.

df = session.sql("select id, parent_id from sample_product_data where id < 10")
# Because the underlying SQL statement for the DataFrame is a SELECT statement,
# you can call the filter method to transform this DataFrame.
results = df.filter(col("id") < 3).select(col("id")).collect()
# Prepend a return statement to return the collect() results in a Python worksheet

# In this example, the underlying SQL statement is not a SELECT statement.
df = session.sql("ls @my_stage")
# Calling the filter method results in an error.
try:
  df.filter(col("size") > 50).collect()
except SnowparkSQLException as e:
  print(e.message)
Copy
000904 (42000): SQL compilation error: error line 1 at position 104
invalid identifier 'SIZE'

Envio de consultas ao Snowpark simultaneamente

Nota

Esse recurso requer a versão 1.24 ou superior da biblioteca Snowpark para Python e a versão 8.46 ou superior do servidor.

Os objetos de sessão thread-safe permitem que diferentes partes do código do Snowpark Python sejam executadas simultaneamente enquanto usam a mesma sessão. Isso permite que várias operações - como transformações em vários DataFrames - sejam executadas simultaneamente. Isso é especialmente útil quando você está trabalhando com consultas que podem ser processadas independentemente no servidor Snowflake e se alinha com uma abordagem de multithreading mais tradicional.

O Global Interpreter Lock (GIL) em Python é um mutex que protege o acesso aos objetos Python, impedindo que vários threads nativos executem o bytecode Python simultaneamente. Embora as operações vinculadas a E/S ainda possam se beneficiar do modelo de threading do Python devido ao fato de o GIL ser liberado durante as operações de E/S, os threads vinculados a CPU não atingirão o verdadeiro paralelismo porque somente um thread pode ser executado por vez.

Além disso, quando usado dentro do Snowflake (por exemplo, em um procedimento armazenado), o servidor Python do Snowpark gerencia o Global Interpreter Lock (GIL), liberando-o antes de enviar as consultas ao Snowflake. Isso garante que a verdadeira simultaneidade possa ser obtida ao enfileirar várias consultas de threads separados. Com esse gerenciamento, o Snowpark permite que vários threads enviem consultas simultaneamente, garantindo a execução paralela ideal.

Benefícios do uso de objetos de sessão Thread-Safe no Snowpark

A capacidade de executar várias operações do DataFrame ao mesmo tempo pode trazer os seguintes benefícios aos usuários do Snowpark:

  • Desempenho aprimorado: os objetos de sessão thread-safe permitem que você execute várias consultas do Snowpark Python ao mesmo tempo, reduzindo o tempo de execução geral. Por exemplo, se você precisar processar várias tabelas de forma independente, esse recurso reduzirá significativamente o tempo necessário para concluir o trabalho, pois não será mais necessário aguardar a conclusão do processamento de cada tabela antes de iniciar a próxima.

  • Utilização eficiente da computação: o envio de consultas simultâneas garante que os recursos de computação do Snowflake sejam usados de forma eficiente, reduzindo os tempos ociosos.

  • Usabilidade: os objetos de sessão thread-safe integram-se perfeitamente ao multithreading nativo das Python APIs, o que permite que os desenvolvedores aproveitem as ferramentas internas do Python para controlar o comportamento do thread e otimizar a execução paralela.

Os objetos de sessão thread-safe e os trabalhos assíncronos podem se complementar, dependendo do seu caso de uso. Os trabalhos assíncronos são úteis quando o usuário não precisa esperar que os trabalhos terminem, permitindo a execução sem bloqueio e sem gerenciamento de pool de threads. Os objetos de sessão thread-safe, por outro lado, são úteis quando você deseja enviar várias consultas simultaneamente do lado do cliente. Em alguns casos, os blocos de código também podem conter trabalhos assíncronos, permitindo que ambos os métodos sejam usados juntos de forma eficaz.

A seguir, alguns exemplos em que os objetos de sessão thread-safe podem aprimorar seu pipeline de dados.

Exemplo 1: carregamento simultâneo de várias tabelas

Este exemplo demonstra o carregamento de dados de três arquivos CSV diferentes em três tabelas separadas, usando três threads para executar o comando COPY INTO simultaneamente.

import threading
from snowflake.snowpark import Session

# Define the list of tables
tables = ["customers", "orders", "products"]

# Function to copy data from stage to tables
def execute_copy(table_name):
    try:
        # Read data from the stage using DataFrameReader
        df = (
            session.read.option("SKIP_HEADER", 1)
            .option("PATTERN", f"{table_name}[.]csv")
            .option("FORCE", True)
            .csv(f"@my_stage")
        )

        # Copy data into the target table
        df.copy_into_table(
            table_name=table_name, target_columns=session.table(table_name).columns
        )

    except Exception as e:
        print(f"Failed to copy data into {table_name}, Error: {e}")

# Create an empty list of threads
threads = []

# Loop through and start a thread for each table
for table in tables:
    thread = threading.Thread(target=execute_copy, args=(table,))
    threads.append(thread)
    thread.start()

# Wait for all threads to finish
for thread in threads:
    thread.join()
Copy

Exemplo 2: processamento simultâneo de várias tabelas

Este exemplo demonstra como você pode usar vários threads para filtrar, agregar e inserir simultaneamente dados em uma tabela de resultados de cada tabela de transação do cliente (transaction_customer1, transaction_customer2 e transaction_customer3).

from concurrent.futures import ThreadPoolExecutor
from snowflake.snowpark import Session
from snowflake.snowpark.functions import col, month, sum, lit

# List of customers
customers = ["customer1", "customer2", "customer3"]

# Define a function to process each customer transaction table
def process_customer_table(customer_name):
    table_name = f"transaction_{customer_name}"

    try:
        # Load the customer transaction table
        df = session.table(table_name)
        print(f"Processing {table_name}...")

        # Filter data by positive values and non null categories
        df_filtered = df.filter((col("value") > 0) & col("category").is_not_null())

        # Perform aggregation: Sum of value by category and month
        df_aggregated = df_filtered.with_column("month", month(col("date"))).with_column("customer_name", lit(customer_name)).group_by(col("category"), col("month"), col("customer_name")).agg(sum("value").alias("total_value"))

        # Save the processed data into a new result table
        df_aggregated.show()
        df_aggregated.write.save_as_table("aggregate_customers", mode="append")
        print(f"Data from {table_name} processed and saved")

    except Exception as e:
        print(f"Error processing {table_name}: {e}")

# Using ThreadPoolExecutor to handle concurrency
with ThreadPoolExecutor(max_workers=3) as executor:
    # Submit tasks for each customer table
    executor.map(process_customer_table, customers)

# Display the results from the aggregate table
session.table("aggregate_customers").show()
Copy

Limitações do uso de objetos de sessão Thread-Safe

  • Se precisar gerenciar várias transações simultaneamente, é importante usar vários objetos de sessão, pois vários threads de uma única sessão não são compatíveis com transações simultâneas.

  • Alterar as configurações de tempo de execução da sessão (incluindo variáveis de sessão do Snowflake, como banco de dados, esquema, warehouse e configurações do lado do cliente, como cte_optimization_enabled, sql_simplifier_enabled) enquanto outros threads estão ativos pode levar a um comportamento inesperado. Para evitar conflitos, é melhor usar objetos de sessão separados se diferentes threads exigirem configurações distintas. Por exemplo, se você precisar executar operações em diferentes bancos de dados em paralelo, certifique-se de que cada thread tenha seu próprio objeto de sessão em vez de compartilhar a mesma sessão.

Como retornar o conteúdo de um DataFrame como um DataFrame do Pandas

Para retornar o conteúdo de um DataFrame como um DataFrame do Pandas, use o método to_pandas.

Por exemplo:

python_df = session.create_dataframe(["a", "b", "c"])
pandas_df = python_df.to_pandas()
Copy

DataFrames Snowpark vs. DataFrame Snowpark pandas: Qual devo escolher?

Ao instalar a biblioteca Snowpark Python, você tem a opção de usar a DataFrames API ou o pandas on Snowflake.

Os DataFrames Snowpark foram modelados a partir do PySpark, enquanto o Snowpark pandas foi criado para estender a funcionalidade do DataFrame Snowpark e fornecer uma interface familiar aos usuários do pandas para facilitar a migração e adoção. Recomendamos usar diferentes APIs dependendo do seu caso de uso e preferência:

Use os Snowpark pandas se você…

Use o DataFrames Snowpark se você…

Prefere trabalhar com ou ter código existente escrito em pandas

Prefere trabalhar com ou ter código existente escrito em Spark

Tem um fluxo de trabalho que envolva análise interativa e exploração iterativa

Tem um fluxo de trabalho que envolva processamento em lote e desenvolvimento iterativo limitado

Está acostumado a trabalhar com operações de DataFrame que são executadas imediatamente

Está acostumado a trabalhar com operações de DataFrame que são avaliadas ociosamente

Prefere que os dados sejam consistentes e ordenados durante as operações

Concorda com os dados não serem ordenados?

Concorda em ter um desempenho ligeiramente mais lento em comparação com o DataFrames Snowpark, em favor de uma API mais fácil de usar

Prioriza o desempenho em vez da facilidade de uso

Do ponto de vista da implementação, os DataFrames Snowpark e os DataFrames pandas são semanticamente diferentes. Como os DataFrames Snowpark são modelados a partir do PySpark, eles operam na fonte de dados original, obtêm os dados atualizados mais recentes e, portanto, não mantêm a ordem das operações. Snowpark pandas é modelado a partir do pandas, que opera em um instantâneo dos dados, mantém a ordem durante a operação e permitem indexação posicional baseada na ordem. A manutenção de pedidos é útil para inspeção visual de dados na análise interativa de dados.

Para obter mais informações, consulte Como usar o pandas on Snowflake com o DataFrames Snowpark.

Linguagem: Português