Como trabalhar com DataFrames no Snowpark Java¶

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 que devem ser selecionadas.

Para se referir a uma coluna, crie um objeto Column chamando o método estático Functions.col.

DataFrame dfProductInfo = session.table("sample_product_data").select(Functions.col("id"), Functions.col("name"));
dfProductInfo.show();

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 = 12"
  // in an SQL SELECT statement.
  DataFrame df = session.table("sample_product_data");
  df.filter(Functions.col("id").equal_to(Functions.lit(12))).show();
  

  Copy

  // Specify the equivalent of "WHERE key + category_id < 10"
  // in an SQL SELECT statement.
  DataFrame df2 = session.table("sample_product_data");
  df2.filter(Functions.col("key").plus(Functions.col("category_id")).lt(Functions.lit(10))).show();
  

  Copy

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

  // Specify the equivalent of "SELECT key * 10 AS c"
  // in an SQL SELECT statement.
  DataFrame df3 = session.table("sample_product_data");
  df3.select(Functions.col("key").multiply(Functions.lit(10)).as("c")).show();
  

  Copy

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

  // Specify the equivalent of "sample_a JOIN sample_b on sample_a.id_a = sample_b.id_a"
  // in an SQL SELECT statement.
  DataFrame dfLhs = session.table("sample_a");
  DataFrame dfRhs = session.table("sample_b");
  DataFrame dfJoined = dfLhs.join(dfRhs, dfLhs.col("id_a").equal_to(dfRhs.col("id_a")));
  dfJoined.show();
  

  Copy

// Create a DataFrame that joins two other DataFrames (dfLhs and dfRhs).
// Use the DataFrame.col method to refer to the columns used in the join.
DataFrame dfLhs = session.table("sample_a");
DataFrame dfRhs = session.table("sample_b");
DataFrame dfJoined = dfLhs.join(dfRhs, dfLhs.col("id_a").equal_to(dfRhs.col("id_a"))).select(dfLhs.col("value").as("L"), dfRhs.col("value").as("R"));
dfJoined.show();

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. Quando você especifica um nome, o Snowflake considera que o nome está em maiúsculas. Por exemplo, as chamadas a seguir são equivalentes:

// The following calls are equivalent:
df.select(Functions.col("id123"));
df.select(Functions.col("ID123"));

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 Scala. 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:

DataFrame df = session.table("\"10tablename\"");

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:

// The following calls are equivalent:
df.select(Functions.col("3rdID"));
df.select(Functions.col("\"3rdID\""));

// The following calls are equivalent:
df.select(Functions.col("id with space"));
df.select(Functions.col("\"id with space\""));

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:

describe table quoted;
+------------------------+ ...
| name                   | ...
|------------------------+ ...
| name_with_"air"_quotes | ...
| "column_name_quoted"   | ...
+------------------------+ ...

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"""):

DataFrame dfTable = session.table("quoted");
dfTable.select("\"name_with_\"\"air\"\"_quotes\"");
dfTable.select("\"\"\"column_name_quoted\"\"\"");

Tenha em mente que 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(Functions.col("id with space"));
df.select(Functions.col("ID WITH SPACE"));

Como usar literais como objetos de coluna¶

Para usar um literal em um método que passa um objeto Column, crie um objeto Column para o literal passando o literal para o método estático lit na classe Functions. Por exemplo:

// Show the first 10 rows in which category_id is greater than 5.
// Use `Functions.lit(5)` to create a Column object for the literal 5.
DataFrame df = session.table("sample_product_data");
df.filter(Functions.col("category_id").gt(Functions.lit(5))).show();

Se o literal é um ponto flutuante ou valor duplo em Java (por exemplo, 0.05 é tratado como Double por padrão), a biblioteca do Snowpark gera um SQL que implicitamente converte o valor para o tipo de dados correspondente do Snowpark (por exemplo, 0.05::DOUBLE). Isso pode produzir um valor aproximado que difira do número exato especificado.

Por exemplo, o código a seguir não exibe linhas correspondentes, mesmo que o filtro (que corresponde a valores maiores ou iguais a 0.05) deva corresponder às linhas do DataFrame:

// Create a DataFrame that contains the value 0.05.
DataFrame df = session.sql("select 0.05 :: Numeric(5, 2) as a");

// Applying this filter results in no matching rows in the DataFrame.
df.filter(Functions.col("a").leq(Functions.lit(0.06).minus(Functions.lit(0.01)))).show();

O problema é que Functions.lit(0.06) e Functions.lit(0.01) produzem valores aproximados para 0.06 e 0.01, não os valores exatos.

Para evitar esse problema, converta o literal para o tipo Snowpark que você deseja usar. Por exemplo, para usar um NUMBER com uma precisão de 5 e escala de 2:

import com.snowflake.snowpark_java.types.*;
...

df.filter(Functions.col("a").leq(Functions.lit(0.06).cast(DataTypes.createDecimalType(5, 2)).minus(Functions.lit(0.01).cast(DataTypes.createDecimalType(5, 2))))).show();

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 tipo com.snowflake.snowpark_java.types package. Por exemplo, para converter um literal para um NUMBER com uma precisão de 5 e escala de 2:

// Import for the DecimalType class..
import com.snowflake.snowpark_java.types.*;

Column decimalValue = Functions.lit(0.05).cast(DataTypes.createDecimalType(5,2));

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.

DataFrame dfProductInfo = session.table("sample_product_data").filter(Functions.col("id").equal_to(Functions.lit(1))).select(Functions.col("name"), Functions.col("serial_number"));
dfProductInfo.show();

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(Functions.col("id").equal_to(Functions.lit(1))) retorna um DataFrame para a tabela sample_product_data que está preparada para retornar a linha com id = 1.

  Observe novamente 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(Functions.col("name"), Functions.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.

Quando você fizer chamadas a métodos encadeadas, tenha em mente que a ordem das chamadas é importante. Cada chamada de método retorna um DataFrame que foi transformado. Certifique-se de que as chamadas subsequentes funcionem com os DataFrame transformados.

Por exemplo, no código abaixo, o método select retorna um DataFrame que contém apenas duas colunas: name e serial_number. A chamada ao método filter nesse DataFrame falha porque usa a coluna id, que não está no DataFrame transformado.

// This fails with the error "invalid identifier 'ID'."
DataFrame dfProductInfo = session.table("sample_product_data").select(Functions.col("name"), Functions.col("serial_number")).filter(Functions.col("id").equal_to(Functions.lit(1)));
dfProductInfo.show();

Em contraste, o seguinte código é executado com sucesso porque o método filter() é chamado em um DataFrame que contém todas as colunas da tabela sample_product_data (incluindo a coluna id):

// This succeeds because the DataFrame returned by the table() method
// includes the "id" column.
DataFrame dfProductInfo = session.table("sample_product_data").filter(Functions.col("id").equal_to(Functions.lit(1))).select(Functions.col("name"), Functions.col("serial_number"));
dfProductInfo.show();

Tenha em mente que 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 limitar o número de linhas em um DataFrame¶

Para limitar o número de linhas em um DataFrame, você pode usar o método de transformação limit.

A API do Snowpark também fornece métodos de ação para recuperar e imprimir um número limitado de linhas:

• o método de ação first (para executar a consulta e retornar as primeiras n linhas)

• o método de ação show (para executar a consulta e imprimir as primeiras n linhas)

Esses métodos efetivamente adicionam uma cláusula LIMIT à instrução SQL que é executada.

Como explicado nas notas de uso para LIMIT, os resultados não são determinísticos, a menos que você especifique uma ordem de classificação (ORDER BY) em conjunto com LIMIT.

Para manter a cláusula ORDER BY com a cláusula LIMIT (por exemplo, para que ORDER BY não esteja em uma subconsulta separada), você deve chamar o método que limita os resultados do DataFrame retornado pelo método sort.

Por exemplo, se você estiver encadeando chamadas a métodos:

DataFrame df = session.table("sample_product_data");

// Limit the number of rows to 5, sorted by parent_id.
DataFrame dfSubset = df.sort(Functions.col("parent_id")).limit(5);

// Return the first 5 rows, sorted by parent_id.
Row[] arrayOfRows = df.sort(Functions.col("parent_id")).first(5);

// Print the first 5 rows, sorted by parent_id.
df.sort(Functions.col("parent_id")).show(5);

Como obter definições das colunas¶

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

import com.snowflake.snowpark_java.types.*;
...

// Get the StructType object that describes the columns in the
// underlying rowset.
StructType tableSchema = session.table("sample_product_data").schema();
System.out.println("Schema for sample_product_data: " + tableSchema);

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 o método schema e depois chama o método names no objeto retornado StructType para obter uma array de nomes de colunas. Os nomes são normalizados no método StructType retornado pelo método schema.

import java.util.Arrays;
...

// Create a DataFrame containing the "id" and "3rd" columns.
DataFrame dfSelectedColumns = session.table("sample_product_data").select(Functions.col("id"), Functions.col("3rd"));
// Print out the names of the columns in the schema.
System.out.println(Arrays.toString(dfSelectedColumns.schema().names()));

Junção de DataFrames¶

Para unir objetos DataFrame, chame o método join.

As seções a seguir explicam como usar DataFrames para realizar uma junção:

• Configuração dos dados de exemplo para as junções

• Como especificar as colunas para a junção

• Como realizar uma junção natural

• Como especificar o tipo de junção

• Junção de tabelas múltiplas

• Como realizar uma autojunção

CREATE OR REPLACE TABLE sample_a (
  id_a INTEGER,
  name_a VARCHAR,
  value INTEGER
);
INSERT INTO sample_a (id_a, name_a, value) VALUES
  (10, 'A1', 5),
  (40, 'A2', 10),
  (80, 'A3', 15),
  (90, 'A4', 20)
;
CREATE OR REPLACE TABLE sample_b (
  id_b INTEGER,
  name_b VARCHAR,
  id_a INTEGER,
  value INTEGER
);
INSERT INTO sample_b (id_b, name_b, id_a, value) VALUES
  (4000, 'B1', 40, 10),
  (4001, 'B2', 10, 5),
  (9000, 'B3', 80, 15),
  (9099, 'B4', NULL, 200)
;
CREATE OR REPLACE TABLE sample_c (
  id_c INTEGER,
  name_c VARCHAR,
  id_a INTEGER,
  id_b INTEGER
);
INSERT INTO sample_c (id_c, name_c, id_a, id_b) VALUES
  (1012, 'C1', 10, NULL),
  (1040, 'C2', 40, 4000),
  (1041, 'C3', 40, 4001)
;

// Create a DataFrame that joins the DataFrames for the tables
// "sample_a" and "sample_b" on the column named "id_a".
DataFrame dfLhs = session.table("sample_a");
DataFrame dfRhs = session.table("sample_b");
DataFrame dfJoined = dfLhs.join(dfRhs, dfLhs.col("id_a").equal_to(dfRhs.col("id_a")));
dfJoined.show();

----------------------------------------------------------------------
|"ID_A"  |"NAME_A"  |"VALUE"  |"ID_B"  |"NAME_B"  |"ID_A"  |"VALUE"  |
----------------------------------------------------------------------
|10      |A1        |5        |4001    |B2        |10      |5        |
|40      |A2        |10       |4000    |B1        |40      |10       |
|80      |A3        |15       |9000    |B3        |80      |15       |
----------------------------------------------------------------------

DataFrame dfLhs = session.table("sample_a");
DataFrame dfRhs = session.table("sample_b");
DataFrame dfJoined = dfLhs.join(dfRhs, dfLhs.col("id_a").equal_to(dfRhs.col("id_a")));
DataFrame dfSelected = dfJoined.select(dfLhs.col("value").as("LeftValue"), dfRhs.col("value").as("RightValue"));
dfSelected.show();

------------------------------
|"LEFTVALUE"  |"RIGHTVALUE"  |
------------------------------
|5            |5             |
|10           |10            |
|15           |15            |
------------------------------

--------------------------------------------------------------------------------------------------
|"l_ZSz7_ID_A"  |"NAME_A"  |"l_ZSz7_VALUE"  |"ID_B"  |"NAME_B"  |"r_heec_ID_A"  |"r_heec_VALUE"  |
--------------------------------------------------------------------------------------------------
|10             |A1        |5               |4001    |B2        |10             |5               |
|40             |A2        |10              |4000    |B1        |40             |10              |
|80             |A3        |15              |9000    |B3        |80             |15              |
--------------------------------------------------------------------------------------------------

DataFrame dfLhs = session.table("sample_a");
DataFrame dfRhs = session.table("sample_b");
DataFrame dfJoined = dfLhs.naturalJoin(dfRhs);
dfJoined.show();

---------------------------------------------------
|"ID_A"  |"VALUE"  |"NAME_A"  |"ID_B"  |"NAME_B"  |
---------------------------------------------------
|10      |5        |A1        |4001    |B2        |
|40      |10       |A2        |4000    |B1        |
|80      |15       |A3        |9000    |B3        |
---------------------------------------------------

// Create a DataFrame that performs a left outer join on
// "sample_a" and "sample_b" on the column named "id_a".
DataFrame dfLhs = session.table("sample_a");
DataFrame dfRhs = session.table("sample_b");
DataFrame dfLeftOuterJoin = dfLhs.join(dfRhs, dfLhs.col("id_a").equal_to(dfRhs.col("id_a")), "left");
dfLeftOuterJoin.show();

----------------------------------------------------------------------
|"ID_A"  |"NAME_A"  |"VALUE"  |"ID_B"  |"NAME_B"  |"ID_A"  |"VALUE"  |
----------------------------------------------------------------------
|40      |A2        |10       |4000    |B1        |40      |10       |
|10      |A1        |5        |4001    |B2        |10      |5        |
|80      |A3        |15       |9000    |B3        |80      |15       |
|90      |A4        |20       |NULL    |NULL      |NULL    |NULL     |
----------------------------------------------------------------------

DataFrame dfFirst = session.table("sample_a");
DataFrame dfSecond  = session.table("sample_b");
DataFrame dfThird = session.table("sample_c");
DataFrame dfJoinThreeTables = dfFirst.join(dfSecond, dfFirst.col("id_a").equal_to(dfSecond.col("id_a"))).join(dfThird, dfFirst.col("id_a").equal_to(dfThird.col("id_a")));
dfJoinThreeTables.show();

------------------------------------------------------------------------------------------------------------
|"ID_A"  |"NAME_A"  |"VALUE"  |"ID_B"  |"NAME_B"  |"ID_A"  |"VALUE"  |"ID_C"  |"NAME_C"  |"ID_A"  |"ID_B"  |
------------------------------------------------------------------------------------------------------------
|10      |A1        |5        |4001    |B2        |10      |5        |1012    |C1        |10      |NULL    |
|40      |A2        |10       |4000    |B1        |40      |10       |1040    |C2        |40      |4000    |
|40      |A2        |10       |4000    |B1        |40      |10       |1041    |C3        |40      |4001    |
------------------------------------------------------------------------------------------------------------

// This fails because columns named "id" and "parent_id"
// are in the left and right DataFrames in the join.
DataFrame df = session.table("sample_product_data");
DataFrame dfJoined = df.join(df, Functions.col("id").equal_to(Functions.col("parent_id")));

// This fails because columns named "id" and "parent_id"
// are in the left and right DataFrames in the join.
DataFrame df = session.table("sample_product_data");
DataFrame dfJoined = df.join(df, df.col("id").equal_to(df.col("parent_id")));

Exception in thread "main" com.snowflake.snowpark_java.SnowparkClientException:
  Joining a DataFrame to itself can lead to incorrect results due to ambiguity of column references.
  Instead, join this DataFrame to a clone() of itself.

// Create a DataFrame object for the "sample_product_data" table for the left-hand side of the join.
DataFrame dfLhs = session.table("sample_product_data");
// Clone the DataFrame object to use as the right-hand side of the join.
DataFrame dfRhs = dfLhs.clone();

// Create a DataFrame that joins the two DataFrames
// for the "sample_product_data" table on the
// "id" and "parent_id" columns.
DataFrame dfJoined = dfLhs.join(dfRhs, dfLhs.col("id").equal_to(dfRhs.col("parent_id")));
dfJoined.show();

// Create a DataFrame that performs a self-join on a DataFrame
// using the column named "key".
DataFrame df = session.table("sample_product_data");
DataFrame dfJoined = df.join(df, "key");

Como executar uma ação de forma síncrona¶

Para executar uma ação de forma síncrona, chame um dos seguintes métodos de ação:

Por exemplo, 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.
DataFrame dfProducts = session.table("sample_product_data");

// Send the query to the server for execution and
// print the count of rows in the table.
System.out.println("Rows returned: " + dfProducts.count());

Você também pode chamar métodos de ação para:

• Executar uma consulta em uma tabela e retornar os resultados.

• Executar uma consulta e imprimir os resultados no console.

Nota: Se você estiver chamando o método schema para obter as definições das colunas no DataFrame, você não precisa chamar um método de ação.

Como executar uma ação de forma assíncrona¶

Para executar uma ação de forma assíncrona, chame o método async para retornar um objeto “ator assíncrono” (por exemplo, DataFrameAsyncActor) e chamar um método de ação assíncrona nesse objeto.

Esses métodos de ação de um objeto de ator assíncrono retornam um objeto TypedAsyncJob, que você pode usar para verificar o status da ação assíncrona e obter os resultados da ação.

As próximas seções explicam como executar as ações de forma assíncrona e verificar os resultados.

• Explicação do fluxo básico de ações assíncronas

• Como especificar o número máximo de segundos a esperar

• Como acessar uma consulta assíncrona por ID

import java.util.Arrays;

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

// Execute the query asynchronously.
// This call does not block.
TypedAsyncJob<Row[]> asyncJob = df.async().collect();
// Check if the query has completed execution.
System.out.println("Is query " + asyncJob.getQueryId() + " done? " + asyncJob.isDone());
// Get an Array of Rows containing the results, and print the results.
// Note that getResult is a blocking call.
Row[] results = asyncJob.getResult();
System.out.println(Arrays.toString(results));

// Create a DataFrame for the "sample_product_data" table.
DataFrame dfProducts = session.table("sample_product_data");

// Execute the query asynchronously.
// This call does not block.
TypedAsyncJob<Long> asyncJob = dfProducts.async().count();
// Check if the query has completed execution.
System.out.println("Is query " + asyncJob.getQueryId() + " done? " + asyncJob.isDone());
// Print the count of rows in the table.
// Note that getResult is a blocking call.
System.out.println("Rows returned: " + asyncJob.getResult());

// Wait a maximum of 10 seconds for the query to complete before retrieving the results.
Row[] results = asyncJob.getResult(10);

import java.util.Arrays;
...

AsyncJob asyncJob = session.createAsyncJob(myQueryId);
// Check if the query has completed execution.
System.out.println("Is query " + asyncJob.getQueryId() + " done? " + asyncJob.isDone());
// If you need to retrieve the results, call getRows to return an Array of Rows containing the results.
// Note that getRows is a blocking call.
Row[] rows = asyncJob.getRows();
System.out.println(Arrays.toString(rows));

Como retornar todas as linhas¶

Para retornar todas as linhas de uma vez, chame o método collect. Esse método retorna uma array de objetos Row. Para obter os valores da linha, chame o método getType (por exemplo, getString, getInt, etc.).

Por exemplo:

Row[] rows = session.table("sample_product_data").select(Functions.col("name"), Functions.col("category_id")).sort(Functions.col("name")).collect();
for (Row row : rows) {
  System.out.println("Name: " + row.getString(0) + "; Category ID: " + row.getInt(1));
}

Como retornar um iterador para as linhas¶

Se você quiser usar um Iterator para iterar sobre os objetos Row nos resultados, chame toLocalIterator. Se a quantidade de dados nos resultados for grande, o método carregará as linhas por parte para evitar carregar todas as linhas de uma só vez na memória.

Por exemplo:

import java.util.Iterator;

Iterator<Row> rowIterator = session.table("sample_product_data").select(Functions.col("name"), Functions.col("category_id")).sort(Functions.col("name")).toLocalIterator();
while (rowIterator.hasNext()) {
  Row row = rowIterator.next();
  System.out.println("Name: " + row.getString(0) + "; Category ID: " + row.getInt(1));
}

Como retornar as primeiras n linhas¶

Para retornar as primeiras n linhas, chame o método first passando o número de linhas a serem retornadas.

Como explicado em Como limitar o número de linhas em um DataFrame, os resultados são não determinísticos. Se você quiser que os resultados sejam determinísticos, chame esse método em um DataFrame ordenado (df.sort().first()).

Por exemplo:

import java.util.Arrays;
...

DataFrame df = session.table("sample_product_data");
Row[] rows = df.sort(Functions.col("name")).first(5);
System.out.println(Arrays.toString(rows));

Atualização de linhas em uma tabela¶

Para atualizar as linhas em uma tabela, chame o método update ou updateColumn, passando um Map que associa as colunas a serem atualizadas e os valores correspondentes a serem atribuídos a essas colunas:

• Para especificar os nomes das colunas como cadeias de caracteres no Map, chame updateColumn.

• Para especificar objetos Column no Map, chame update.

Ambos os métodos retornam um objeto UpdateResult contendo o número de linhas que foram atualizadas. (Consulte UpdateResult.)

Por exemplo, para substituir os valores na coluna chamada count pelo valor 1, se você deseja usar um Map que associa o nome da coluna (a String) com o valor correspondente, chame updateColumn:

import java.util.HashMap;
import java.util.Map;
...

Map<String, Column> assignments = new HashMap<>();
assignments.put("3rd", Functions.lit(1));
Updatable updatableDf = session.table("sample_product_data");
UpdateResult updateResult = updatableDf.updateColumn(assignments);
System.out.println("Number of rows updated: " + updateResult.getRowsUpdated());

Se você quiser usar um objeto Column no Map para identificar a coluna a ser atualizada, chame update:

import java.util.HashMap;
import java.util.Map;
...

Map<Column, Column> assignments = new HashMap<>();
assignments.put(Functions.col("3rd"), Functions.lit(1));
Updatable updatableDf = session.table("sample_product_data");
UpdateResult updateResult = updatableDf.update(assignments);
System.out.println("Number of rows updated: " + updateResult.getRowsUpdated());

Se a atualização só deve ser feita quando uma condição for cumprida, você pode especificar essa condição como um argumento. Por exemplo, para substituir os valores na coluna chamada count por 2 para linhas nas quais a coluna category_id tem o valor 20:

import java.util.HashMap;
import java.util.Map;
...
Map<Column, Column> assignments = new HashMap<>();
assignments.put(Functions.col("3rd"), Functions.lit(2));
Updatable updatableDf = session.table("sample_product_data");
UpdateResult updateResult = updatableDf.update(assignments, Functions.col("category_id").equal_to(Functions.lit(20)));
System.out.println("Number of rows updated: " + updateResult.getRowsUpdated());

Se você precisar basear a condição em uma junção com um objeto DataFrame diferente, você pode passar esse DataFrame como argumento e usar o DataFrame na condição. Por exemplo, para substituir os valores na coluna chamada count por 3 para linhas nas quais a coluna category_id corresponde à category_id no DataFrame dfParts:

import java.util.HashMap;
import java.util.Map;
...
Map<Column, Column> assignments = new HashMap<>();
assignments.put(Functions.col("3rd"), Functions.lit(3));
Updatable updatableDf = session.table("sample_product_data");
DataFrame dfParts = session.table("parts");
UpdateResult updateResult = updatableDf.update(assignments, updatableDf.col("category_id").equal_to(dfParts.col("category_id")), dfParts);
System.out.println("Number of rows updated: " + updateResult.getRowsUpdated());

Como excluir linhas de uma tabela¶

Para o método delete, você pode especificar uma condição que identifica as linhas a serem excluídas, e pode basear essa condição em uma junção com outro DataFrame. delete retorna um objeto DeleteResult, contendo o número de linhas que foram excluídas. (Consulte DeleteResult.)

Por exemplo, para apagar as linhas que têm o valor 1 na coluna category_id:

Updatable updatableDf = session.table("sample_product_data");
DeleteResult deleteResult = updatableDf.delete(updatableDf.col("category_id").equal_to(Functions.lit(1)));
System.out.println("Number of rows deleted: " + deleteResult.getRowsDeleted());

Se a condição se refere a colunas em um DataFrame diferente, passe esse DataFrame como o segundo argumento. Por exemplo, para excluir as linhas nas quais a coluna category_id corresponde à category_id no DataFrame dfParts, passe dfParts como o segundo argumento:

Updatable updatableDf = session.table("sample_product_data");
DeleteResult deleteResult = updatableDf.delete(updatableDf.col("category_id").equal_to(dfParts.col("category_id")), dfParts);
System.out.println("Number of rows deleted: " + deleteResult.getRowsDeleted());

Fusão de linhas em uma tabela¶

Para inserir, atualizar e excluir linhas em uma tabela com base nos valores de uma segunda tabela ou subconsulta (o equivalente ao comando MERGE em SQL), faça o seguinte:

1. No objeto Updatable para a tabela onde você quer fundir os dados, chame o método merge, passando o objeto DataFrame para a outra tabela e a expressão da coluna para a condição de junção.

   Isso retorna um objeto MergeBuilder que você pode usar para especificar as ações a serem tomadas (por exemplo, inserir, atualizar ou excluir) nas linhas que correspondem e nas que não correspondem aos critérios. (Consulte MergeBuilder.)

2. Usando o objeto MergeBuilder:

   • Para especificar a atualização ou exclusão que deve ser feita nas linhas correspondentes, chame o método whenMatched.

     Se você precisar especificar uma condição adicional quando as linhas devem ser atualizadas ou excluídas, você pode passar uma expressão de coluna para essa condição.

     Esse método retorna um objeto MatchedClauseBuilder que você pode usar para especificar a ação a ser executada. (Consulte MatchedClauseBuilder.)

     Chame o método update ou delete no objeto MatchedClauseBuilder para especificar a ação de atualização ou exclusão que deve ser executada nas linhas correspondentes. Esses métodos devolvem um objeto MergeBuilder que você pode usar para especificar cláusulas adicionais.

   • Para especificar a inserção que deve ser feita quando as linhas não correspondem, chame o método whenNotMatched.

     Se você precisar especificar uma condição adicional quando as linhas devem ser inseridas, você pode passar uma expressão de coluna para essa condição.

     Esse método retorna um objeto NotMatchedClauseBuilder que você pode usar para especificar a ação a ser executada. (Consulte NotMatchedClauseBuilder.)

     Chame o método insert no objeto NotMatchedClauseBuilder para especificar a ação de inserção que deve ser executada quando as linhas não correspondem. Esses métodos devolvem um objeto MergeBuilder que você pode usar para especificar cláusulas adicionais.

3. Quando terminar de especificar as inserções, atualizações e exclusões que devem ser realizadas, chame o método collect do objeto MergeBuilder para realizar as inserções, atualizações e exclusões especificadas na tabela.

   collect retorna um objeto MergeResult contendo o número de linhas que foram inseridas, atualizadas e excluídas. (Consulte MergeResult.)

O exemplo a seguir insere uma linha com as colunas id e value da tabela source na tabela target se a tabela target não contiver uma linha com uma ID correspondente:

MergeResult mergeResult = target.merge(source, target.col("id").equal_to(source.col("id")))
                    .whenNotMatched().insert([source.col("id"), source.col("value")])
                    .collect();

O exemplo a seguir atualiza uma linha na tabela target com o valor da coluna value da linha na tabela source que tem a mesma ID:

import java.util.HashMap;
import java.util.Map;
...
Map<String, Column> assignments = new HashMap<>();
assignments.put("value", source.col("value"));
MergeResult mergeResult = target.merge(source, target.col("id").equal_to(source.col("id")))
                    .whenMatched().update(assignments)
                    .collect();

Carregamento e descarregamento de arquivos em um estágio¶

Para carregar e baixar arquivos em um estágio, use os métodos put e get do objeto FileOperation:

• Carregamento de arquivos para um estágio

• Como baixar arquivos de um estágio

Como usar fluxos de entrada para carregar e descarregar dados em um estágio¶

Para usar fluxos de entrada para carregar dados em um arquivo em um estágio e baixar dados de um arquivo em um estágio, use os métodos uploadStream e downloadStream do objeto FileOperation:

• Como usar um fluxo de entrada para carregar dados em um arquivo em um estágio

• Como usar um fluxo de entrada para baixar dados de um arquivo em um estágio

import java.io.InputStream;
...
boolean compressData = true;
String pathToFileOnStage = "@myStage/path/file";
session.file().uploadStream(pathToFileOnStage, new ByteArrayInputStream(fileContent.getBytes()), compressData);

import java.io.InputStream;
...
boolean isDataCompressed = true;
String pathToFileOnStage = "@myStage/path/file";
InputStream is = session.file().downloadStream(pathToFileOnStage, isDataCompressed);

Como configurar um DataFrame para arquivos em um estágio¶

Esta seção explica como configurar um DataFrame para arquivos em um estágio do Snowflake. Após criar esse DataFrame, você pode usar o DataFrame para:

• obter dados dos arquivos

• copiar dados dos arquivos em uma tabela

Para configurar um DataFrame para arquivos em um estágio do Snowflake, use a classe DataFrameReader:

1. Verifique se você tem os seguintes privilégios:

   • Privilégios de acesso aos arquivos no estágio.

   • Um dos seguintes:

     • privilégios CREATE TABLE no esquema, se você planeja especificar opções de cópia que determinam como os dados são copiados dos arquivos preparados.

     • Em outros casos, privilégios CREATE FILE FORMAT sobre o esquema.

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

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

   1. Crie um objeto StructType que consiste em uma array de objetos StructField que descrevem 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 pacote com.snowflake.snowpark_java.types).

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

      Por exemplo:

      import com.snowflake.snowpark_java.types.*;
      ...
      
      StructType schemaForDataFile = StructType.create(
        new StructField("id", DataTypes.StringType, true),
        new StructField("name", DataTypes.StringType, true));
      

      Copy

   3. Chame o método schema no objeto DataFrameReader, passando o objeto StructType.

      Por exemplo:

      DataFrameReader dfReader = session.read().schema(schemaForDataFile);
      

      Copy

      O método 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.

4. 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 o método DataFrameReader.option ou o método DataFrameReader.options.

   Passe o nome e valor da opção que você deseja definir. Você pode definir os seguintes tipos de opções:

   • As opções de formato de arquivo descritas na documentação em CREATE FILE FORMAT.

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

   dfReader = dfReader.option("field_delimiter", ";").option("COMPRESSION", "NONE");
   

   Copy

   O método option retorna um objeto DataFrameReader que é configurado com as opções especificadas.

   Para definir várias opções, você pode encadear chamadas ao método option (como mostrado no exemplo acima) ou chamar o método DataFrameReader.options, passando um Map dos nomes e valores das opções.

5. Chame o método correspondente ao formato dos arquivos. Você pode chamar um dos seguintes métodos:

   • DataFrameReader.avro

   • DataFrameReader.csv

   • DataFrameReader.json

   • DataFrameReader.orc

   • DataFrameReader.parquet

   • DataFrameReader.xml

   Ao chamar esses métodos, passe o local do estágio dos arquivos a serem lidos. Por exemplo:

   DataFrame df = dfReader.csv("@mystage/myfile.csv");
   

   Copy

   Para especificar vários arquivos que começam com o mesmo prefixo, especifique o prefixo após o nome do estágio. Por exemplo, para carregar arquivos que tenham o prefixo csv_ do estágio @mystage:

   DataFrame df = dfReader.csv("@mystage/csv_");
   

   Copy

   Os métodos correspondentes ao formato de um arquivo retornam um objeto CopyableDataFrame para esse arquivo. CopyableDataFrame estende DataFrame e fornece métodos adicionais para trabalhar os dados em arquivos preparados.

6. Chame um método de ação para:

   • obter dados dos arquivos, ou

   • copiar dados dos arquivos em uma tabela

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

Carregamento de dados de arquivos em um DataFrame¶

Depois que você configurar um DataFrame para arquivos em um estágio, você pode carregar dados dos arquivos para o DataFrame:

1. 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 chamado data.json no estágio chamado mystage:

   DataFrame df = session.read().json("@mystage/data.json").select(Functions.col("$1").subField("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.

2. Chame o método DataFrame.collect para carregar os dados. Por exemplo:

   Row[] results = df.collect();
   

   Copy

Como copiar dados de arquivos para uma tabela¶

Depois de configurar um DataFrame para arquivos em um estágio, você pode chamar o método copyInto para copiar os dados em uma tabela. Esse método executa o comando COPY INTO <tabela>.

Por exemplo, o seguinte código carrega os dados do arquivo CSV especificado por myFileStage para a tabela mytable. Como os dados estão em um arquivo CSV, o código também deve descrever os campos no arquivo. O exemplo faz isso chamando o método schema do objeto DataFrameReader e passando um objeto StructType (schemaForDataFile) contendo uma array de objetos StructField que descrevem os campos.

CopyableDataFrame copyableDf = session.read().schema(schemaForDataFile).csv("@mystage/myfile.csv");
copyableDf.copyInto("mytable");

Como salvar um DataFrame para arquivos em um estágio¶

Se você precisar salvar um DataFrame para arquivos em um estágio, você pode chamar o método DataFrameWriter correspondente ao formato do arquivo (por exemplo, o método csv para escrever em um arquivo CSV), passando o local do estágio onde os arquivos devem ser salvos. Esses métodos DataFrameWriter executam o comando COPY INTO <local>.

Para salvar o conteúdo de um DataFrame em arquivos em um estágio:

1. Chame o método write do objeto DataFrame para obter um objeto DataFrameWriter. Por exemplo, para obter o objeto DataFrameWriter para um DataFrame que representa a tabela chamada sample_product_data:

   DataFrameWriter dfWriter = session.table("sample_product_data").write();
   

   Copy

2. Se você quiser substituir o conteúdo do arquivo (se o arquivo existir), chame o método mode do objeto DataFrameWriter, passando SaveMode.Overwrite.

   Caso contrário, por padrão, o DataFrameWriter reporta um erro se o arquivo especificado no estágio já existir.

   O método mode retorna o mesmo objeto DataFrameWriter configurado com o modo especificado.

   Por exemplo, para especificar que o DataFrameWriter deve substituir o arquivo no estágio:

   dfWriter = dfWriter.mode(SaveMode.Overwrite);
   

   Copy

3. Se você precisar especificar informações adicionais sobre como os dados devem ser salvos (por exemplo, que os dados devem ser compactados ou que você deseja usar um ponto-e-vírgula para delimitar campos em um arquivo CSV), chame o método DataFrameWriter.option ou o método DataFrameWriter.options.

   Passe o nome e valor da opção que você deseja definir. Você pode definir os seguintes tipos de opções:

   • As opções de formato de arquivo descritas na documentação em COPY INTO <location>.

   • As opções de cópia descritas na documentação em COPY INTO <location>.

   • PARTITION BY ou HEADER.

   Observe que você não pode usar o método option para definir as seguintes opções:

   • A opção do tipo de formato TYPE.

   • A opção de cópia OVERWRITE. Em vez disso, para definir esta opção, chame o método mode (como mencionado na etapa anterior).

   O exemplo a seguir configura o objeto DataFrameWriter para salvar dados em um arquivo CSV sem compressão, usando um ponto-e-vírgula (no lugar de uma vírgula) como delimitador de campo.

   dfWriter = dfWriter.option("field_delimiter", ";").option("COMPRESSION", "NONE");
   

   Copy

   O método option retorna um objeto DataFrameWriter que é configurado com a opção especificada.

   Para definir várias opções, você pode encadear chamadas ao método option (como mostrado no exemplo acima) ou chamar o método DataFrameWriter.options, passando um Map dos nomes e valores das opções.

4. Para retornar detalhes sobre cada arquivo que foi salvo, defina a DETAILED_OUTPUT opção de cópia como TRUE.

   Por padrão, DETAILED_OUTPUT é FALSE, o que significa que o método retorna uma única linha de saída contendo os campos "rows_unloaded", "input_bytes" e "output_bytes".

   Quando você define DETAILED_OUTPUT como TRUE, o método retorna uma linha de saída para cada arquivo salvo. Cada linha contém os campos FILE_NAME, FILE_SIZE e ROW_COUNT.

5. Chame o método correspondente ao formato do arquivo para salvar os dados no arquivo. Você pode chamar um dos seguintes métodos:

   • DataFrameWriter.csv

   • DataFrameWriter.json

   • DataFrameWriter.parquet

   Ao chamar esses métodos, passe o local do estágio do arquivo onde os dados devem ser escritos (por exemplo, @mystage).

   Por padrão, o método salva os dados em nomes de arquivos com o prefixo data_ (por exemplo, @mystage/data_0_0_0.csv). Se você quiser que os arquivos sejam nomeados com um prefixo diferente, especifique o prefixo após o nome do estágio. Por exemplo:

   WriteFileResult writeFileResult = dfWriter.csv("@mystage/saved_data");
   

   Copy

   Esse exemplo salva o conteúdo do DataFrame em arquivos que começam com o prefixo saved_data (por exemplo, @mystage/saved_data_0_0_0.csv).

6. Verifique o objeto WriteFileResult retornado para obter informações sobre a quantidade de dados escritos no arquivo.

   A partir do objeto WriteFileResult, você pode acessar a saída produzida pelo comando COPY INTO <location>:

   • Para acessar as linhas de saída como uma array de objetos Row, chame o método getRows.

   • Para determinar quais campos estão presentes nas linhas, chame o método getSchema, que retorna um StructType que descreve os campos na linha.

   Por exemplo, para imprimir os nomes dos campos e valores nas linhas de saída:

   WriteFileResult writeFileResult = dfWriter.csv("@mystage/saved_data");
   Row[] rows = writeFileResult.getRows();
   StructType schema = writeFileResult.getSchema();
   for (int i = 0 ; i < rows.length ; i++) {
     System.out.println("Row:" + i);
     Row row = rows[i];
     for (int j = 0; j < schema.size(); j++) {
       System.out.println(schema.get(j).name() + ": " + row.get(j));
     }
   }
   

   Copy

O exemplo a seguir usa um DataFrame para salvar o conteúdo da tabela chamada car_sales para arquivos JSON com o prefixo saved_data no estágio @mystage (por exemplo, @mystage/saved_data_0_0_0.json). O código de exemplo:

• Substitui o arquivo se ele já existir no estágio.

• Retorna resultados detalhados sobre a operação de salvar.

• Salva os dados sem compressão.

Finalmente, o código de exemplo imprime cada campo e valor nas linhas de saída retornadas:

DataFrame df = session.table("car_sales");
WriteFileResult writeFileResult = df.write().mode(SaveMode.Overwrite).option("DETAILED_OUTPUT", "TRUE").option("compression", "none").json("@mystage/saved_data");
Row[] rows = writeFileResult.getRows();
StructType schema = writeFileResult.getSchema();
for (int i = 0 ; i < rows.length ; i++) {
  System.out.println("Row:" + i);
  Row row = rows[i];
  for (int j = 0; j < schema.size(); j++) {
    System.out.println(schema.get(j).name() + ": " + row.get(j));
  }
}

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 subField(«<field_name>») para retornar um objeto Column para um campo em um OBJECT (ou uma VARIANT que contenha um OBJECT).

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

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

DataFrame df = session.table("car_sales");
df.select(Functions.col("src").subField("dealership")).show();

O código imprime a seguinte saída:

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

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:

DataFrame df = session.table("car_sales");
df.select(Functions.col("src").subField("salesperson").subField("name")).show();

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.

DataFrame df = session.table("car_sales");
df.select(Functions.col("src").subField("vehicle").subField(0)).show();
df.select(Functions.col("src").subField("vehicle").subField(0).subField("price")).show();

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 ao método apply, você pode usar as funções Functions.get, Functions.get_ignore_case ou Functions.get_path se o nome do campo ou elementos no caminho forem irregulares e dificultarem o uso dos métodos Column.subField.

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

df.select(Functions.get(Functions.col("src"), Functions.lit("dealership"))).show();
df.select(Functions.col("src").subField("dealership")).show();

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

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

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.
import com.snowflake.snowpark_java.types.*;
...
DataFrame df = session.table("car_sales");
df.select(Functions.col("src").subField("salesperson").subField("id")).show();
df.select(Functions.col("src").subField("salesperson").subField("id").cast(DataTypes.StringType)).show();

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 de uma array), chame o método flatten. 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:

DataFrame df = session.table("car_sales");
df.flatten(Functions.col("src").subField("customer")).show();