Working with DataFrames in Snowpark Scala¶

Specifying Columns and Expressions¶

When calling these transformation methods, you might need to specify columns or expressions that use columns. For example, when calling the select method, you need to specify the columns that should be selected.

To refer to a column, create a Column object by calling the col function in the com.snowflake.snowpark.functions object.

// Import the col function from the functions object.
import com.snowflake.snowpark.functions._

val dfProductInfo = session.table("sample_product_data").select(col("id"), col("name"))
dfProductInfo.show()

When specifying a filter, projection, join condition, etc., you can use Column objects in an expression. For example:

• You can use Column objects with the filter method to specify a filter condition:

  // Specify the equivalent of "WHERE id = 20"
  // in an SQL SELECT statement.
  df.filter(col("id") === 20)
  

  Copy

  // Specify the equivalent of "WHERE a + b < 10"
  // in an SQL SELECT statement.
  df.filter((col("a") + col("b")) < 10)
  

  Copy

• You can use Column objects with the select method to define an alias:

  // Specify the equivalent of "SELECT b * 10 AS c"
  // in an SQL SELECT statement.
  df.select((col("b") * 10) as "c")
  

  Copy

• You can use Column objects with the join method to define a join condition:

  // Specify the equivalent of "X JOIN Y on X.a_in_X = Y.b_in_Y"
  // in an SQL SELECT statement.
  dfX.join(dfY, col("a_in_X") === col("b_in_Y"))
  

  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.
val dfJoined = dfLhs.join(dfRhs, dfLhs.col("key") === dfRhs.col("key")).select(dfLhs.col("value").as("L"), dfRhs.col("value").as("R"))

// Create a DataFrame that joins two other DataFrames (dfLhs and dfRhs).
// Use the DataFrame.apply method to refer to the columns used in the join.
// Note that dfLhs("key") is shorthand for dfLhs.apply("key").
val dfJoined = dfLhs.join(dfRhs, dfLhs("key") === dfRhs("key")).select(dfLhs("value").as("L"), dfRhs("value").as("R"))

Using Shorthand For a Column Object¶

As an alternative to using the col function, you can refer to a column in one of these ways:

• Use a dollar sign in front of the quoted column name ($"column_name").

• Use an apostrophe (a single quote) in front of the unquoted column name ('column_name).

To do this, import the names from the implicits object after you create a Session object:

val session = Session.builder.configFile("/path/to/properties").create

// Import this after you create the session.
import session.implicits._

// Use the $ (dollar sign) shorthand.
val df = session.table("T").filter($"id" === 10).filter(($"a" + $"b") < 10)

// Use ' (apostrophe) shorthand.
val df = session.table("T").filter('id === 10).filter(('a + 'b) < 10).select('b * 10)

Using Double Quotes Around Object Identifiers (Table Names, Column Names, etc.)¶

The names of databases, schemas, tables, and stages that you specify must conform to the Snowflake identifier requirements. When you specify a name, Snowflake considers the name to be in upper case. For example, the following calls are equivalent:

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

If the name does not conform to the identifier requirements, you must use double quotes (") around the name. Use a backslash (\) to escape the double quote character within a Scala string literal. For example, the following table name does not start with a letter or an underscore, so you must use double quotes around the name:

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

Note that when specifying the name of a column, you don’t need to use double quotes around the name. The Snowpark library automatically encloses the column name in double quotes for you if the name does not comply with the identifier requirements:.

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

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

If you have already added double quotes around a column name, the library does not insert additional double quotes around the name.

In some cases, the column name might contain double quote characters:

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

As explained in Identifier requirements, for each double quote character within a double-quoted identifier, you must use two double quote characters (e.g. "name_with_""air""_quotes" and """column_name_quoted"""):

val dfTable = session.table("quoted")
dfTable.select("\"name_with_\"\"air\"\"_quotes\"").show()
dfTable.select("\"\"\"column_name_quoted\"\"\"").show()

Keep in mind that when an identifier is enclosed in double quotes (whether you explicitly added the quotes or the library added the quotes for you), Snowflake treats the identifier as case-sensitive:

// 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"))
df.select(col("ID WITH SPACE"))

Using Literals as Column Objects¶

To use a literal in a method that passes in a Column object, create a Column object for the literal by passing the literal to the lit function in the com.snowflake.snowpark.functions object. For example:

// Import for the lit and col functions.
import com.snowflake.snowpark.functions._

// 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.filter(col("num_items").gt(lit(5))).show()

If the literal is a floating point or double value in Scala (e.g. 0.05 is treated as a Double by default (https://docs.scala-lang.org/overviews/scala-book/built-in-types.html)), the Snowpark library generates SQL that implicitly casts the value to the corresponding Snowpark data type (e.g. 0.05::DOUBLE). This can produce an approximate value that differs from the exact number specified.

For example, the following code displays no matching rows, even though the filter (that matches values greater than or equal to 0.05) should match the rows in the DataFrame:

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

// Applying this filter results in no matching rows in the DataFrame.
df.filter(col("a") <= lit(0.06) - lit(0.01)).show()

The problem is that lit(0.06) and lit(0.01) produce approximate values for 0.06 and 0.01, not the exact values.

To avoid this problem, you can use one of the following approaches:

• Option 1: Cast the literal to the Snowpark type that you want to use. For example, to use a NUMBER with a precision of 5 and a scale of 2:

  df.filter(col("a") <= lit(0.06).cast(new DecimalType(5, 2)) - lit(0.01).cast(new DecimalType(5, 2))).show()
  

  Copy

• Option 2: Cast the value to the type that you want to use before passing the value to the lit function. For example, if you want to use the BigDecimal type (https://docs.scala-lang.org/overviews/scala-book/built-in-types.html#bigint-and-bigdecimal):

  df.filter(col("a") <= lit(BigDecimal(0.06)) - lit(BigDecimal(0.01))).show()
  

  Copy

Casting a Column Object to a Specific Type¶

To cast a Column object to a specific type, call the Column.cast method, and pass in a type object from the com.snowflake.snowpark.types package. For example, to cast a literal as a NUMBER with a precision of 5 and a scale of 2:

// Import for the lit function.
import com.snowflake.snowpark.functions._
// Import for the DecimalType class..
import com.snowflake.snowpark.types._

val decimalValue = lit(0.05).cast(new DecimalType(5,2))

Chaining Method Calls¶

Because each method that transforms a DataFrame object returns a new DataFrame object that has the transformation applied, you can chain method calls (link removed) to produce a new DataFrame that is transformed in additional ways.

The following example returns a DataFrame that is configured to:

• Query the sample_product_data table.

• Return the row with id = 1.

• Select the name and serial_number columns.

val dfProductInfo = session.table("sample_product_data").filter(col("id") === 1).select(col("name"), col("serial_number"))
dfProductInfo.show()

In this example:

• session.table("sample_product_data") returns a DataFrame for the sample_product_data table.

  Although the DataFrame does not yet contain the data from the table, the object does contain the definitions of the columns in the table.

• filter(col("id") === 1) returns a DataFrame for the sample_product_data table that is set up to return the row with id = 1.

  Note again that the DataFrame does not yet contain the matching row from the table. The matching row is not retrieved until you call an action method.

• select(col("name"), col("serial_number")) returns a DataFrame that contains the name and serial_number columns for the row in the sample_product_data table that has id = 1.

When you chain method calls, keep in mind that the order of calls is important. Each method call returns a DataFrame that has been transformed. Make sure that subsequent calls work with the transformed DataFrame.

For example, in the code below, the select method returns a DataFrame that just contains two columns: name and serial_number. The filter method call on this DataFrame fails because it uses the id column, which is not in the transformed DataFrame.

// This fails with the error "invalid identifier 'ID'."
val dfProductInfo = session.table("sample_product_data").select(col("name"), col("serial_number")).filter(col("id") === 1)

In contrast, the following code executes successfully because the filter() method is called on a DataFrame that contains all of the columns in the sample_product_data table (including the id column):

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

Keep in mind that you might need to make the select and filter method calls in a different order than you would use the equivalent keywords (SELECT and WHERE) in a SQL statement.

Limiting the Number of Rows in a DataFrame¶

To limit the number of rows in a DataFrame, you can use the DataFrame.limit transformation method.

The Snowpark API also provides action methods for retrieving and printing out a limited number of rows:

• the DataFrame.first action method (to execute the query and return the first n rows)

• the DataFrame.show action method (to execute the query and print the first n rows)

These methods effectively add a LIMIT clause to the SQL statement that is executed.

As explained in the usage notes for LIMIT, the results are non-deterministic unless you specify a sort order (ORDER BY) in conjunction with LIMIT.

To keep the ORDER BY clause with the LIMIT clause (e.g. so that ORDER BY is not in a separate subquery), you must call the method that limits results on the DataFrame returned by the sort method.

For example, if you are chaining method calls:

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

// Return the first 5 rows, sorted by parent_id.
var arrayOfRows = df.sort(col("parent_id")).first(5)

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

Retrieving Column Definitions¶

To retrieve the definition of the columns in the dataset for the DataFrame, call the schema method. This method returns a StructType object that contains an Array of StructField objects. Each StructField object contains the definition of a column.

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

In the returned StructType object, the column names are always normalized. Unquoted identifiers are returned in uppercase, and quoted identifiers are returned in the exact case in which they were defined.

The following example creates a DataFrame containing the columns named ID and 3rd. For the column name 3rd, the Snowpark library automatically encloses the name in double quotes ("3rd") because the name does not comply with the requirements for an identifier.

The example calls the schema method and then calls the names method on the returned StructType object to get an ArraySeq of column names. The names are normalized in the StructType returned by the schema method.

// Create a DataFrame containing the "id" and "3rd" columns.
val dfSelectedColumns = session.table("sample_product_data").select(col("id"), col("3rd"))
// Print out the names of the columns in the schema. This prints out:
//   ArraySeq(ID, "3rd")
println(dfSelectedColumns.schema.names.toSeq)

Joining DataFrames¶

To join DataFrame objects, call the DataFrame.join method.

The following sections explain how to use DataFrames to perform a join:

• Setting up the Sample Data for the Joins

• Specifying the Columns for the Join

• Performing a Natural Join

• Specifying the Type of Join

• Joining Multiple Tables

• Performing a Self-Join

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".
val dfLhs = session.table("sample_a")
val dfRhs = session.table("sample_b")
val dfJoined = dfLhs.join(dfRhs, dfLhs.col("id_a") === 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       |
----------------------------------------------------------------------

val dfLhs = session.table("sample_a")
val dfRhs = session.table("sample_b")
val dfJoined = dfLhs.join(dfRhs, dfLhs.col("id_a") === dfRhs.col("id_a"))
val 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              |
--------------------------------------------------------------------------------------------------

val dfLhs = session.table("sample_a")
val dfRhs = session.table("sample_b")
val 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".
val dfLhs = session.table("sample_a")
val dfRhs = session.table("sample_b")
val dfLeftOuterJoin = dfLhs.join(dfRhs, dfLhs.col("id_a") === 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     |
----------------------------------------------------------------------

val dfFirst = session.table("sample_a")
val dfSecond  = session.table("sample_b")
val dfThird = session.table("sample_c")
val dfJoinThreeTables = dfFirst.join(dfSecond, dfFirst.col("id_a") === dfSecond.col("id_a")).join(dfThird, dfFirst.col("id_a") === 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.
val df = session.table("sample_product_data");
val dfJoined = df.join(df, col("id") === col("parent_id"))

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

Exception in thread "main" com.snowflake.snowpark.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.
val dfLhs = session.table("sample_product_data")
// Clone the DataFrame object to use as the right-hand side of the join.
val dfRhs = dfLhs.clone()

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

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

Performing an Action Synchronously¶

To perform an action synchronously, call one of the following action methods:

To execute the query and return the number of results, call the count method:

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

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

You can also call action methods to:

• Execute a query against a table and return the results.

• Execute a query and print the results to the console.

Note: If you are calling the schema method to get the definitions of the columns in the DataFrame, you do not need to call an action method.

Performing an Action Asynchronously¶

To perform an action asynchronously, call the async method to return an “async actor” object (e.g. DataFrameAsyncActor), and call an asynchronous action method in that object.

These action methods of an async actor object return a TypedAsyncJob object, which you can use to check the status of the asynchronous action and retrieve the results of the action.

The next sections explain how to perform actions asynchronously and check the results.

• Understanding the Basic Flow of Asynchronous Actions

• Specifying the Maximum Number of Seconds to Wait

• Accessing an Asynchronous Query by ID

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

// Execute the query asynchronously.
// This call does not block.
val asyncJob = df.async.collect()
// Check if the query has completed execution.
println(s"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.
val results = asyncJob.getResult()
results.foreach(println)

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

// Execute the query asynchronously.
// This call does not block.
val asyncJob = df.async.count()
// Check if the query has completed execution.
println(s"Is query ${asyncJob.getQueryId()} done? ${asyncJob.isDone()}")
// Print the count of rows in the table.
// Note that getResult is a blocking call.
println("Rows returned: " + asyncJob.getResult())

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

val asyncJob = session.createAsyncJob(myQueryId)
// Check if the query has completed execution.
println(s"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.
val rows = asyncJob.getRows()
rows.foreach(println)

Returning All Rows¶

To return all rows at once, call the DataFrame.collect method. This method returns an Array of Row objects. To retrieve the values from the row, call the getType method (e.g. getString, getInt, etc.).

For example:

import com.snowflake.snowpark.functions_

val rows = session.table("sample_product_data").select(col("name"), col("category_id")).sort(col("name")).collect()
for (row <- rows) {
  println(s"Name: ${row.getString(0)}; Category ID: ${row.getInt(1)}")
}

Returning an Iterator for the Rows¶

If you want to use an Iterator (https://docs.scala-lang.org/overviews/collections/iterators.html) to iterate over the Row objects in the results, call DataFrame.toLocalIterator. If the amount of data in the results is large, the method loads the rows by chunk to avoid loading all rows into memory at once.

For example:

import com.snowflake.snowpark.functions_

while (rowIterator.hasNext) {
  val row = rowIterator.next()
  println(s"Name: ${row.getString(0)}; Category ID: ${row.getInt(1)}")
}

Returning the First n Rows¶

To return the first n rows, call the DataFrame.first method, passing in the number of rows to return.

As explained in Limiting the Number of Rows in a DataFrame, the results are non-deterministic. If you want the results to be deterministic, call this method on a sorted DataFrame (df.sort().first()).

For example:

import com.snowflake.snowpark.functions_

val df = session.table("sample_product_data")
val rows = df.sort(col("name")).first(5)
rows.foreach(println)

Updating Rows in a Table¶

For the update method, pass in a Map that associates the columns to update and the corresponding values to assign to those columns. update returns an UpdateResult object, which contains the number of rows that were updated. (See UpdateResult.)

For example, to replace the values in the column named count with the value 1:

val updatableDf = session.table("sample_product_data")
val updateResult = updatableDf.update(Map("count" -> lit(1)))
println(s"Number of rows updated: ${updateResult.rowsUpdated}")

The example above uses the name of the column to identify the column. You can also use a column expression:

val updateResult = updatableDf.update(Map(col("count") -> lit(1)))

If the update should be made only when a condition is met, you can specify that condition as an argument. For example, to replace the values in the column named count for rows in which the category_id column has the value 20:

val updateResult = updatableDf.update(Map(col("count") -> lit(1)), col("category_id") === 20)

If you need to base the condition on a join with a different DataFrame object, you can pass that DataFrame in as an argument and use that DataFrame in the condition. For example, to replace the values in the column named count for rows in which the category_id column matches the category_id in the DataFrame dfParts:

val updatableDf = session.table("sample_product_data")
val dfParts = session.table("parts")
val updateResult = updatableDf.update(Map(col("count") -> lit(1)), updatableDf("category_id") === dfParts("category_id"), dfParts)

Deleting Rows in a Table¶

For the delete method, you can specify a condition that identifies the rows to delete, and you can base that condition on a join with another DataFrame. delete returns a DeleteResult object, which contains the number of rows that were deleted. (See DeleteResult.)

For example, to delete the rows that have the value 1 in the category_id column:

val updatableDf = session.table("sample_product_data")
val deleteResult = updatableDf.delete(updatableDf("category_id") === 1)
println(s"Number of rows deleted: ${deleteResult.rowsDeleted}")

If the condition refers to columns in a different DataFrame, pass that DataFrame in as the second argument. For example, to delete the rows in which the category_id column matches the category_id in the DataFrame dfParts, pass in dfParts as the second argument:

val updatableDf = session.table("sample_product_data")
val deleteResult = updatableDf.delete(updatableDf("category_id") === dfParts("category_id"), dfParts)
println(s"Number of rows deleted: ${deleteResult.rowsDeleted}")

Merging Rows into a Table¶

To insert, update, and deletes rows in one table based on values in a second table or a subquery (the equivalent of the MERGE command in SQL), do the following:

1. In the Updatable object for the table where you want the data merged in, call the merge method, passing in the DataFrame object for the other table and the column expression for the join condition.

   This returns a MergeBuilder object that you can use to specify the actions to take (e.g. insert, update, or delete) on the rows that match and the rows that don’t match. (See MergeBuilder.)

2. Using the MergeBuilder object:

   • To specify the update or deletion that should be performed on matching rows, call the whenMatched method.

     If you need to specify an additional condition whe rows should be updated or deleted, you can pass in a column expression for that condition.

     This method returns a MatchedClauseBuilder object that you can use to specify the action to perform. (See MatchedClauseBuilder.)

     Call the update or delete method in the MatchedClauseBuilder object to specify the update or delete action that should be performed on matching rows. These methods return a MergeBuilder object that you can use to specify additional clauses.

   • To specify the insert that should be performed when rows do not match, call the whenNotMatched method.

     If you need to specify an additional condition when rows should be inserted, you can pass in a column expression for that condition.

     This method returns a NotMatchedClauseBuilder object that you can use to specify the action to perform. (See NotMatchedClauseBuilder.)

     Call the insert method in the NotMatchedClauseBuilder object to specify the insert action that should be performed when rows do not match. These methods return a MergeBuilder object that you can use to specify additional clauses.

3. When you are done specifying the inserts, updates, and deletions that should be performed, call the collect method of the MergeBuilder object to perform the specified inserts, updates, and deletions on the table.

   collect returns a MergeResult object, which contains the number of rows that were inserted, updated, and deleted. (See MergeResult.)

The following example inserts a row with the id and value columns from the source table into the target table if the target table does not contain a row with a matching ID:

val mergeResult = target.merge(source, target("id") === source("id"))
                      .whenNotMatched.insert(Seq(source("id"), source("value")))
                      .collect()

The following example updates a row in the target table with the value of the value column from the row in the source table that has the same ID:

val mergeResult = target.merge(source, target("id") === source("id"))
                      .whenMatched.update(Map("value" -> source("value")))
                      .collect()

Uploading and Downloading Files in a Stage¶

To upload and download files in a stage, use the FileOperation object:

• Uploading Files to a Stage

• Downloading Files from a Stage

Using Input Streams to Upload and Download Data in a Stage¶

To use input streams to upload data to a file on a stage and download data from a file on a stage, use the uploadStream and downloadStream methods of the FileOperation object:

• Using an Input Stream to Upload Data to a File on a Stage

• Using an Input Stream to Download Data from a File on a Stage

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

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

Setting Up a DataFrame for Files in a Stage¶

This section explains how to set up a DataFrame for files in a Snowflake stage. Once you create this DataFrame, you can use the DataFrame to:

• retrieve data from the files

• copy data from the files into a table

To set up a DataFrame for files in a Snowflake stage, use the DataFrameReader class:

1. Verify that you have the following privileges:

   • Privileges to access files in the stage.

   • One of the following:

     • CREATE TABLE privileges on the schema, if you plan to specify copy options that determine how data is copied from the staged files.

     • CREATE FILE FORMAT privileges on the schema, otherwise.

2. Call the read method in the Session class to access a DataFrameReader object.

3. If the files are in CSV format, describe the fields in the file. To do this:

   1. Create a StructType object that consists of a sequence of StructField objects that describe the fields in the file.

   2. For each StructField object, specify the following:

      • The name of the field.

      • The data type of the field (specified as an object in the com.snowflake.snowpark.types package).

      • Whether or not the field is nullable.

      For example:

      import com.snowflake.snowpark.types._
      
      val schemaForDataFile = StructType(
          Seq(
              StructField("id", StringType, true),
              StructField("name", StringType, true)))
      

      Copy

   3. Call the schema method in the DataFrameReader object, passing in the StructType object.

      For example:

      var dfReader = session.read.schema(schemaForDataFile)
      

      Copy

      The schema method returns a DataFrameReader object that is configured to read files containing the specified fields.

      Note that you do not need to do this for files in other formats (such as JSON). For those files, the DataFrameReader treats the data as a single field of the VARIANT type with the field name $1.

4. If you need to specify additional information about how the data should be read (for example, that the data is compressed or that a CSV file uses a semicolon instead of a comma to delimit fields), call the DataFrameReader.option method or the DataFrameReader.options method.

   Pass in the name and value of the option that you want to set. You can set the following types of options:

   • The file format options described in the documentation on CREATE FILE FORMAT.

   • The copy options described in the COPY INTO TABLE documentation.

     Note that setting copy options can result in a more expensive execution strategy when you retrieve the data into the DataFrame.

   The following example sets up the DataFrameReader object to query data in a CSV file that is not compressed and that uses a semicolon for the field delimiter.

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

   Copy

   The option method returns a DataFrameReader object that is configured with the specified option.

   To set multiple options, you can either chain calls to the option method (as shown in the example above) or call the DataFrameReader.options method, passing in a Map of the names and values of the options.

5. Call the method corresponding to the format of the files. You can call one of the following methods:

   • DataFrameReader.avro

   • DataFrameReader.csv

   • DataFrameReader.json

   • DataFrameReader.orc

   • DataFrameReader.parquet

   • DataFrameReader.xml

   When calling these methods, pass in the stage location of the files to be read. For example:

   val df = dfReader.csv("@s3_ts_stage/emails/data_0_0_0.csv")
   

   Copy

   To specify multiple files that start with the same prefix, specify the prefix after the stage name. For example, to load files that have the prefix csv_ from the stage @mystage:

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

   Copy

   The methods corresponding to the format of a file return a CopyableDataFrame object for that file. CopyableDataFrame extends DataFrame and provides additional methods for working the data in staged files.

6. Call an action method to:

   • retrieve data from the files, or

   • copy data from the files into a table

   As is the case with DataFrames for tables, the data is not retrieved into the DataFrame until you call an action method.

Loading Data from Files into a DataFrame¶

After you set up a DataFrame for files in a stage, you can load data from the files into the DataFrame:

1. Use the DataFrame object methods to perform any transformations needed on the dataset (for example, selecting specific fields, filtering rows, etc.).

   For example, to extract the color element from a JSON file named data.json in the stage named mystage:

   val df = session.read.json("@mystage/data.json").select(col("$1")("color"))
   

   Copy

   As explained earlier, for files in formats other than CSV (e.g. JSON), the DataFrameReader treats the data in the file as a single VARIANT column with the name $1.

2. Call the DataFrame.collect method to load the data. For example:

   val results = df.collect()
   

   Copy

Copying Data from Files into a Table¶

After you set up a DataFrame for files in a stage, you can call the CopyableDataFrame.copyInto method to copy the data into a table. This method executes the COPY INTO <table> command.

For example, the following code loads data from the CSV file specified by myFileStage into the table mytable. Because the data is in a CSV file, the code must also describe the fields in the file. The example does this by calling the DataFrameReader.schema method and passing in a StructType object (csvFileSchema) containing a sequence of StructField objects that describe the fields.

val df = session.read.schema(csvFileSchema).csv(myFileStage)
df.copyInto("mytable")

Saving a DataFrame to Files on a Stage¶

If you need to save a DataFrame to files on a stage, you can call the DataFrameWriter method corresponding to the format of the file (e.g. the csv method to write to a CSV file), passing in the stage location where the files should be saved. These DataFrameWriter methods execute the COPY INTO <location> command.

To save the contents of a DataFrame to files on a stage:

1. Call the DataFrame.write method to get a DataFrameWriter object. For example, to get the DataFrameWriter object for a DataFrame that represents the table named sample_product_data:

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

   Copy

2. If you want to overwrite the contents of the file (if the file exists), call the DataFrameWriter.mode method, passing in SaveMode.Overwrite.

   Otherwise, by default, the DataFrameWriter reports an error if the specified file on the stage already exists.

   The mode method returns the same DataFrameWriter object configured with the specified mode.

   For example, to specify that the DataFrameWriter should overwrite the file on the stage:

   dfWriter = dfWriter.mode(SaveMode.Overwrite)
   

   Copy

3. If you need to specify additional information about how the data should be saved (for example, that the data should be compressed or that you want to use a semicolon to delimit fields in a CSV file), call the DataFrameWriter.option method or the DataFrameWriter.options method.

   Pass in the name and value of the option that you want to set. You can set the following types of options:

   • The file format options described in the documentation on COPY INTO <location>.

   • The copy options described in the documentation on COPY INTO <location>.

   • PARTITION BY or HEADER.

   Note that you cannot use the option method to set the following options:

   • The TYPE format type option.

   • The OVERWRITE copy option. To set this option, call the mode method instead (as mentioned in the previous step).

   The following example sets up the DataFrameWriter object to save data to a CSV file in uncompressed form, using a semicolon (rather than a comma) as the field delimiter.

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

   Copy

   The option method returns a DataFrameWriter object that is configured with the specified option.

   To set multiple options, you can chain calls to the option method (as shown in the example above) or call the DataFrameWriter.options method, passing in a Map of the names and values of the options.

4. To return details about each file that was saved, set the DETAILED_OUTPUT copy option to TRUE.

   By default, DETAILED_OUTPUT is FALSE, which means that the method returns a single row of output containing the fields "rows_unloaded", "input_bytes", and "output_bytes".

   When you set DETAILED_OUTPUT to TRUE, the method returns a row of output for each file saved. Each row contains the fields FILE_NAME, FILE_SIZE, and ROW_COUNT.

5. Call the method corresponding to the format of the file to save the data to the file. You can call one of the following methods:

   • DataFrameWriter.csv

   • DataFrameWriter.json

   • DataFrameWriter.parquet

   When calling these methods, pass in the stage location of the file where the data should be written (e.g. @mystage).

   By default, the method saves the data to filenames with the prefix data_ (e.g. @mystage/data_0_0_0.csv). If you want the files to be named with a different prefix, specify the prefix after the stage name. For example:

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

   Copy

   This example saves the contents of the DataFrame to files that begin with the prefix saved_data (e.g. @mystage/saved_data_0_0_0.csv).

6. Check the WriteFileResult object returned for information about the amount of data written to the file.

   From the WriteFileResult object, you can access the output produced by the COPY INTO <location> command:

   • To access the rows of output as an array of Row objects, use the rows value member.

   • To determine which fields are present in the rows, use the schema value member, which is a StructType that describes the fields in the row.

   For example, to print out the names of the fields and values in the output rows:

   val writeFileResult = dfWriter.csv("@mystage/saved_data")
   for ((row, index) <- writeFileResult.rows.zipWithIndex) {
     (writeFileResult.schema.fields, writeFileResult.rows(index).toSeq).zipped.foreach {
       (structField, element) => println(s"${structField.name}: $element")
     }
   }
   

   Copy

The following example uses a DataFrame to save the contents of the table named car_sales to JSON files with the prefix saved_data on the stage @mystage (e.g. @mystage/saved_data_0_0_0.json). The sample code:

• Overwrites the file, if the file already exists on the stage.

• Returns detailed output about the save operation.

• Saves the data uncompressed.

Finally, the sample code prints out each field and value in the output rows returned:

val df = session.table("car_sales")
val writeFileResult = df.write.mode(SaveMode.Overwrite).option("DETAILED_OUTPUT", "TRUE").option("compression", "none").json("@mystage/saved_data")
for ((row, index) <- writeFileResult.rows.zipWithIndex) {
  println(s"Row: $index")
  (writeFileResult.schema.fields, writeFileResult.rows(index).toSeq).zipped.foreach {
    (structField, element) => println(s"${structField.name}: $element")
  }
}

Traversing Semi-Structured Data¶

To refer to a specific field or element in semi-structured data, use the following methods of the Column object:

• Use Column.apply(“<field_name>”) to return a Column object for a field in an OBJECT (or a VARIANT that contains an OBJECT).

• Use Column.apply(<index>) to return a Column object for an element in an ARRAY (or a VARIANT that contains an ARRAY).

As mentioned in Using the apply Method to Refer to a Column, you can omit the method name apply:

col("column_name")("field_name")
col("column_name")(index)

For example, the following code selects the dealership field in objects in the src column of the sample data:

val df = session.table("car_sales")
df.select(col("src")("dealership")).show()

The code prints the following output:

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

You can also chain method calls to traverse a path to a specific field or element.

For example, the following code selects the name field in the salesperson object:

val df = session.table("car_sales")
df.select(col("src")("salesperson")("name")).show()

The code prints the following output:

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

As another example, the following code selects the first element of vehicle field, which holds an array of vehicles. The example also selects the price field from the first element.

val df = session.table("car_sales")
df.select(col("src")("vehicle")(0)).show()
df.select(col("src")("vehicle")(0)("price")).show()

The code prints the following output:

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

As an alternative to the apply method, you can use the get, get_ignore_case, or get_path functions if the field name or elements in the path are irregular and make it difficult to use the Column.apply methods.

For example, the following lines of code both print the value of a specified field in an object:

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

Similarly, the following lines of code both print the value of a field at a specified path in an object:

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

Explicitly Casting Values in Semi-Structured Data¶

By default, the values of fields and elements are returned as string literals (including the double quotes), as shown in the examples above.

To avoid unexpected results, call the cast method to cast the value to a specific type. For example, the following code prints out the values without and with casting:

// Import the objects for the data types, including StringType.
import com.snowflake.snowpark.types._
...
val df = session.table("car_sales")
df.select(col("src")("salesperson")("id")).show()
df.select(col("src")("salesperson")("id").cast(StringType)).show()

The code prints the following output:

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

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

Flattening an Array of Objects into Rows¶

If you need to “flatten” semi-structured data into a DataFrame (e.g. producing a row for every object in an array), call the DataFrame.flatten method. This method is equivalent to the FLATTEN SQL function. If you pass in a path to an object or array, the method returns a DataFrame that contains a row for each field or element in the object or array.

For example, in the sample data, src:customer is an array of objects that contain information about a customer. Each object contains a name and address field.

If you pass this path to the flatten function:

val df = session.table("car_sales")
df.flatten(col("src")("customer")).show()

the method returns a 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"                       |       |       |        |         |                                   |                                     |
|    }                                      |       |       |        |         |                                   |                                     |
|  ]                                        |       |       |        |         |                                   |                                     |
|}                                          |       |       |        |         |                                   |                                     |
----------------------------------------------------------------------------------------------------------------------------------------------------------

From this DataFrame, you can select the name and address fields from each object in the VALUE field:

df.flatten(col("src")("customer")).select(col("value")("name"), col("value")("address")).show()

-------------------------------------------------
|"""VALUE""['NAME']"   |"""VALUE""['ADDRESS']"  |
-------------------------------------------------
|"Joyce Ridgely"       |"San Francisco, CA"     |
|"Bradley Greenbloom"  |"New York, NY"          |
-------------------------------------------------

The following code adds to the previous example by casting the values to a specific type and changing the names of the columns:

df.flatten(col("src")("customer")).select(col("value")("name").cast(StringType).as("Customer Name"), col("value")("address").cast(StringType).as("Customer Address")).show()

-------------------------------------------
|"Customer Name"     |"Customer Address"  |
-------------------------------------------
|Joyce Ridgely       |San Francisco, CA   |
|Bradley Greenbloom  |New York, NY        |
-------------------------------------------