快速参考：适用于 SQL 命令的 Snowpark Scala APIs¶
本主题供您快速参考与 SQL 命令对应的一些 Snowpark APIs。
（请注意，此处未列出与 SQL 命令对应的完整 APIs 列表。）
本主题内容：
执行查询¶
选择列¶
要选择特定列，请使用 DataFrame.select。
SQL 语句示例 |
Snowpark 代码示例 |
|---|---|
SELECT id, name FROM sample_product_data;
|
val dfSelectedCols = df.select(col("id"), col("name"))
dfSelectedCols.show()
|
重命名列¶
要重命名列，请使用 Column.as、Column.alias 或 Column.name。
SQL 语句示例 |
Snowpark 代码示例 |
|---|---|
SELECT id AS item_id FROM sample_product_data;
|
val dfRenamedCol = df.select(col("id").as("item_id"))
dfRenamedCol.show()
|
val dfRenamedCol = df.select(col("id").alias("item_id"))
dfRenamedCol.show()
|
|
val dfRenamedCol = df.select(col("id").name("item_id"))
dfRenamedCol.show()
|
筛选数据¶
要筛选数据，请使用 DataFrame.filter 或 DataFrame.where。
SQL 语句示例 |
Snowpark 代码示例 |
|---|---|
SELECT * FROM sample_product_data WHERE id = 1;
|
val dfFilteredRows = df.filter((col("id") === 1))
dfFilteredRows.show()
|
val dfFilteredRows = df.where((col("id") === 1))
dfFilteredRows.show()
|
对数据进行排序¶
要对数据进行排序，请使用 DataFrame.sort。
SQL 语句示例 |
Snowpark 代码示例 |
|---|---|
SELECT * FROM sample_product_data ORDER BY category_id;
|
val dfSorted = df.sort(col("category_id"))
dfSorted.show()
|
限制返回的行数¶
要限制返回的行数，请使用 DataFrame.limit。请参阅 限制 DataFrame 中的行数。
SQL 语句示例 |
Snowpark 代码示例 |
|---|---|
SELECT * FROM sample_product_data
ORDER BY category_id LIMIT 2;
|
val dfSorted = df.sort(col("category_id")).limit(2);
val arrayRows = dfSorted.collect()
|
执行联接¶
要执行联接，请使用 DataFrame.join 或 DataFrame.naturalJoin。请参阅 联接 DataFrames。
SQL 语句示例 |
Snowpark 代码示例 |
|---|---|
SELECT * FROM sample_a
INNER JOIN sample_b
on sample_a.id_a = sample_b.id_a;
|
val dfJoined =
dfLhs.join(dfRhs, dfLhs.col("id_a") === dfRhs.col("id_a"))
dfJoined.show()
|
SELECT * FROM sample_a NATURAL JOIN sample_b;
|
val dfJoined = dfLhs.naturalJoin(dfRhs)
dfJoined.show()
|
查询半结构化数据¶
要遍历半结构化数据，请使用 Column.apply("<field_name>") 和 Column.apply(<index>)。请参阅 使用半结构化数据。
SQL 语句示例 |
Snowpark 代码示例 |
|---|---|
SELECT src:salesperson.name FROM car_sales;
|
dfJsonField =
df.select(col("src")("salesperson")("name"))
dfJsonField.show()
|
对数据进行分组和聚合¶
要对数据进行分组，请使用 DataFrame.groupBy。这将返回 RelationalGroupedDataFrame 对象，您可以使用该对象执行聚合。
SQL 语句示例 |
Snowpark 代码示例 |
|---|---|
SELECT category_id, count(*)
FROM sample_product_data GROUP BY category_id;
|
val dfCountPerCategory = df.groupBy(col("category")).count()
dfCountPerCategory.show()
|
调用窗口函数¶
要调用:doc:窗口函数</user-guide/functions-window-using>，请使用 Window 对象方法来构建 WindowSpec 对象，该对象反过来可用于窗口函数（类似于使用 '<function> OVER ...PARTITION BY ...ORDER BY'）。
SQL 语句示例 |
Snowpark 代码示例 |
|---|---|
SELECT category_id, price_date, SUM(amount) OVER
(PARTITION BY category_id ORDER BY price_date)
FROM prices ORDER BY price_date;
|
val window = Window.partitionBy(
col("category")).orderBy(col("price_date"))
val dfCumulativePrices = dfPrices.select(
col("category"), col("price_date"),
sum(col("amount")).over(window)).sort(col("price_date"))
dfCumulativePrices.show()
|
更新、删除和合并行¶
要更新、删除和合并表中的行，请使用 Updatable。请参阅 更新、删除和合并表中的行。
SQL 语句示例 |
Snowpark 代码示例 |
|---|---|
UPDATE sample_product_data
SET serial_number = 'xyz' WHERE id = 12;
|
val updateResult =
updatableDf.update(
Map("serial_number" -> lit("xyz")),
col("id") === 12)
|
DELETE FROM sample_product_data
WHERE category_id = 50;
|
val deleteResult =
updatableDf.delete(updatableDf("category_id") === 50)
|
MERGE INTO target_table USING source_table
ON target_table.id = source_table.id
WHEN MATCHED THEN
UPDATE SET target_table.description =
source_table.description;
|
val mergeResult =
target.merge(source, target("id") === source("id"))
.whenMatched.update(Map("description" -> source("description")))
.collect()
|
使用暂存区¶
有关使用暂存区的更多信息，请参阅 处理暂存区中的文件。
从暂存区上传和下载文件¶
要从暂存区上传和下载文件，请使用 FileOperation。请参阅 在暂存区中上传和下载文件。
SQL 语句示例 |
Snowpark 代码示例 |
|---|---|
PUT file:///tmp/*.csv @myStage OVERWRITE = TRUE;
|
val putOptions = Map("OVERWRITE" -> "TRUE")
val putResults = session.file.put(
"file:///tmp/*.csv", "@myStage", putOptions)
|
GET @myStage file:///tmp PATTERN = '.*.csv.gz';
|
val getOptions = Map("PATTERN" -> s"'.*.csv.gz'")
val getResults = session.file.get(
"@myStage", "file:///tmp", getOptions)
|
从暂存区的文件中读取数据¶
要从暂存区的文件中读取数据，请使用 DataFrameReader 为数据创建 DataFrame。请参阅 为暂存区中的文件设置 DataFrame。
SQL 语句示例 |
Snowpark 代码示例 |
|---|---|
CREATE FILE FORMAT snowpark_temp_format TYPE = JSON;
SELECT "$1"[0]['salesperson']['name'] FROM (
SELECT $1::VARIANT AS "$1" FROM @mystage/car_sales.json(
FILE_FORMAT => 'snowpark_temp_format')) LIMIT 10;
DROP FILE FORMAT snowpark_temp_format;
|
val df = session.read.json(
"@mystage/car_sales.json").select(
col("$1")(0)("salesperson")("name"))
df.show();
|
将暂存区文件中的数据复制到表中¶
要将暂存区文件中的数据复制到表中，请使用 DataFrameReader 为数据创建 CopyableDataFrame，并使用 CopyableDataFrame.copyInto 方法将数据复制到表中。请参阅 将数据从文件复制到表中。
SQL 语句示例 |
Snowpark 代码示例 |
|---|---|
COPY INTO new_car_sales
FROM @mystage/car_sales.json
FILE_FORMAT = (TYPE = JSON);
|
val dfCopyableDf = session.read.json("@mystage/car_sales.json")
dfCopyableDf.copyInto("new_car_sales")
|
将 DataFrame 保存到暂存区上的文件¶
要将 DataFrame 保存到暂存区上的文件，请使用以您要用的文件格式命名的 DataFrameWriter 方法。请参阅 将 DataFrame 保存到暂存区上的文件。
SQL 语句示例 |
Snowpark 代码示例 |
|---|---|
COPY INTO @mystage/saved_data.json
FROM ( SELECT * FROM (car_sales) )
FILE_FORMAT = ( TYPE = JSON COMPRESSION = 'none' )
OVERWRITE = TRUE
DETAILED_OUTPUT = TRUE
|
val df = session.table("car_sales")
val writeFileResult = df.write.mode(
SaveMode.Overwrite).option(
"DETAILED_OUTPUT", "TRUE").option(
"compression", "none").json(
"@mystage/saved_data.json")
|
创建和调用用户定义的函数 (UDFs)¶
要创建用作 UDF（匿名 UDF）的 Scala 函数，请使用 udf。
要创建可按名称调用的临时或永久 UDF，请使用 UDFRegistration.registerTemporary 或 UDFRegistration.registerPermanent。
要按名称调用永久 UDF，请使用 callUDF。
有关详细信息，请参阅 在 Scala 中为 DataFrames 创建用户定义的函数 (UDFs) 和 调用标量用户定义的函数 (UDFs)。
SQL 语句示例 |
Snowpark 代码示例 |
|---|---|
CREATE FUNCTION <temp_function_name>
RETURNS INT
LANGUAGE JAVA
...
AS
...;
SELECT ...,
<temp_function_name>(amount) AS doublenum
FROM sample_product_data;
|
val doubleUdf = udf((x: Int) => x + x)
val dfWithDoubleNum = df.withColumn(
"doubleNum", doubleUdf(col("amount")))
dfWithDoubleNum.show()
|
CREATE FUNCTION <temp_function_name>
RETURNS INT
LANGUAGE JAVA
...
AS
...;
SELECT ...,
<temp_function_name>(amount) AS doublenum
FROM sample_product_data;
|
session.udf.registerTemporary(
"doubleUdf", (x: Int) => x + x)
val dfWithDoubleNum = df.withColumn(
"doubleNum", callUDF("doubleUdf", (col("amount"))))
dfWithDoubleNum.show()
|
CREATE FUNCTION doubleUdf(arg1 INT)
RETURNS INT
LANGUAGE JAVA
...
AS
...;
SELECT ...,
doubleUdf(amount) AS doublenum
FROM sample_product_data;
|
session.udf.registerPermanent(
"doubleUdf", (x: Int) => x + x, "mystage")
val dfWithDoubleNum = df.withColumn(
"doubleNum", callUDF("doubleUdf", (col("amount"))))
dfWithDoubleNum.show()
|
创建和调用存储过程¶
有关使用 Snowpark 创建存储过程的指南，请参阅 在 Scala 中为 DataFrames 创建存储过程。
要创建匿名或命名的临时过程，请使用 com.snowflake.snowpark.SProcRegistration 的
registerTemporary方法。要创建命名的永久过程，请使用 com.snowflake.snowpark.SProcRegistration 类的
registerPermanent方法。要调用过程，请使用 com.snowflake.snowpark.Session 类的
storedProcedure方法。
SQL 语句示例 |
Snowpark 代码示例 |
|---|---|
CREATE PROCEDURE <temp_procedure_name>(x INTEGER, y INTEGER)
RETURNS INTEGER
LANGUAGE JAVA
...
AS
$$
BEGIN
RETURN x + y;
END
$$
;
CALL <temp_procedure_name>(2, 3);
|
StoredProcedure sp =
session.sproc().registerTemporary((Session session, Integer x, Integer y) -> x + y,
new DataType[] {DataTypes.IntegerType, DataTypes.IntegerType},
DataTypes.IntegerType);
session.storedProcedure(sp, 2, 3).show();
|
CREATE PROCEDURE sproc(x INTEGER, y INTEGER)
RETURNS INTEGER
LANGUAGE JAVA
...
AS
$$
BEGIN
RETURN x + y;
END
$$
;
CALL sproc(2, 3);
|
String name = "sproc";
StoredProcedure sp =
session.sproc().registerTemporary(name,
(Session session, Integer x, Integer y) -> x + y,
new DataType[] {DataTypes.IntegerType, DataTypes.IntegerType},
DataTypes.IntegerType);
session.storedProcedure(name, 2, 3).show();
|
CREATE PROCEDURE add_hundred(x INTEGER)
RETURNS INTEGER
LANGUAGE JAVA
...
AS
$$
BEGIN
RETURN x + 100;
END
$$
;
CALL add_hundred(3);
|
val name: String = "add_hundred"
val stageName: String = "sproc_libs"
val sp: StoredProcedure =
session.sproc.registerPermanent(
name,
(session: Session, x: Int) => x + 100,
stageName,
true
)
session.storedProcedure(name, 3).show
|