Kurzübersicht: Snowpark Scala-APIs für SQL-Befehle¶
Dieses Thema bietet eine Kurzübersicht einiger Snowpark-APIs, die den SQL-Befehlen entsprechen.
(Beachten Sie, dass dies keine vollständige Liste der APIs ist, die SQL-Befehlen entsprechen).
Unter diesem Thema:
Ausführen von Abfragen¶
Auswählen von Spalten¶
Um bestimmte Spalten auszuwählen, verwenden Sie DataFrame.select.
Beispiel für eine SQL-Anweisung |
Beispiel für Snowpark-Code |
|---|---|
SELECT id, name FROM sample_product_data;
|
val dfSelectedCols = df.select(col("id"), col("name"))
dfSelectedCols.show()
|
Umbenennen von Spalten¶
Verwenden Sie zum Umbenennen einer Spalte Column.as, Column.alias oder Column.name.
Beispiel für eine SQL-Anweisung |
Beispiel für Snowpark-Code |
|---|---|
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()
|
Filtern von Daten¶
Verwenden Sie zum Filtern von Daten DataFrame.filter oder DataFrame.where.
Beispiel für eine SQL-Anweisung |
Beispiel für Snowpark-Code |
|---|---|
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()
|
Sortieren von Daten¶
Verwenden Sie zum Sortieren von Daten DataFrame.sort.
Beispiel für eine SQL-Anweisung |
Beispiel für Snowpark-Code |
|---|---|
SELECT * FROM sample_product_data ORDER BY category_id;
|
val dfSorted = df.sort(col("category_id"))
dfSorted.show()
|
Begrenzen der zurückgegebenen Zeilenanzahl¶
Um die Anzahl der zurückgegebenen Zeilen zu begrenzen, verwenden Sie DataFrame.limit. Siehe Begrenzen der Anzahl von Zeilen in einem DataFrame.
Beispiel für eine SQL-Anweisung |
Beispiel für Snowpark-Code |
|---|---|
SELECT * FROM sample_product_data
ORDER BY category_id LIMIT 2;
|
val dfSorted = df.sort(col("category_id")).limit(2);
val arrayRows = dfSorted.collect()
|
Ausführen von Verknüpfungen (Joins)¶
Verwenden Sie zum Verknüpfen DataFrame.join oder DataFrame.naturalJoin. Siehe Verknüpfen von DataFrames.
Beispiel für eine SQL-Anweisung |
Beispiel für Snowpark-Code |
|---|---|
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()
|
Abfragen von semistrukturierten Daten¶
Verwenden Sie zum Durchlaufen semistrukturierter Daten Column.apply(„<Feldname>“) und Column.apply(<Index>). Siehe Verwenden von semistrukturierten Daten.
Beispiel für eine SQL-Anweisung |
Beispiel für Snowpark-Code |
|---|---|
SELECT src:salesperson.name FROM car_sales;
|
dfJsonField =
df.select(col("src")("salesperson")("name"))
dfJsonField.show()
|
Gruppieren und Aggregieren von Daten¶
Verwenden Sie zum Gruppieren von Daten DataFrame.groupBy. Dies gibt ein RelationalGroupedDataFrame-Objekt zurück, das Sie zum Ausführen von Aggregationen verwenden können.
Beispiel für eine SQL-Anweisung |
Beispiel für Snowpark-Code |
|---|---|
SELECT category_id, count(*)
FROM sample_product_data GROUP BY category_id;
|
val dfCountPerCategory = df.groupBy(col("category")).count()
dfCountPerCategory.show()
|
Aufrufen von Fensterfunktionen¶
Um eine Fensterfunktion aufzurufen, verwenden Sie die Methoden des Window-Objekts, um ein WindowSpec-Objekt zu erstellen, das Sie wiederum für Fensterfunktionen verwenden können (ähnlich wie bei „<Funktion> OVER … PARTITION BY … ORDER BY“).
Beispiel für eine SQL-Anweisung |
Beispiel für Snowpark-Code |
|---|---|
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()
|
Aktualisieren, Löschen und Zusammenführen von Zeilen¶
Verwenden Sie zum Aktualisieren, Löschen und Zusammenzuführen von Zeilen Updatable. Siehe Aktualisieren, Löschen und Zusammenführen von Zeilen in einer Tabelle.
Beispiel für eine SQL-Anweisung |
Beispiel für Snowpark-Code |
|---|---|
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()
|
Verwenden von Stagingbereichen¶
Weitere Informationen zum Verwenden von Stagingbereichen finden Sie unter Verwenden von Dateien in Stagingbereichen.
Hochladen und Herunterladen von Dateien in/aus Stagingbereichen¶
Verwenden Sie zum Hoch- und Herunterladen von Dateien in/aus Stagingbereichen FileOperation. Siehe Hochladen und Herunterladen von Dateien in Stagingbereichen.
Beispiel für eine SQL-Anweisung |
Beispiel für Snowpark-Code |
|---|---|
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)
|
Lesen von Daten aus Dateien in einem Stagingbereich¶
Verwenden Sie zum Lesen von Daten aus Dateien, die sich in einem Stagingbereich befinden, die DataFrameReader-Klasse, die das Erstellen eines DataFrame-Objekts für die Daten ermöglicht. Siehe Einrichten eines DataFrame für Dateien in einem Stagingbereich.
Beispiel für eine SQL-Anweisung |
Beispiel für Snowpark-Code |
|---|---|
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();
|
Kopieren von Daten aus Dateien in einem Stagingbereich in eine Tabelle¶
Verwenden Sie zum Kopieren von Dateien von einem Stagingbereich in eine Tabelle DataFrameReader, um ein CopyableDataFrame-Objekt für die Daten zu erstellen, und verwenden Sie die Methode CopyableDataFrame.copyInto, um die Daten in die Tabelle zu kopieren. Siehe Kopieren von Daten aus Dateien in eine Tabelle.
Beispiel für eine SQL-Anweisung |
Beispiel für Snowpark-Code |
|---|---|
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")
|
Speichern eines DataFrame in Dateien in einem Stagingbereich¶
Um ein DataFrame-Objekt in Dateien in einem Stagingbereich zu speichern, verwenden Sie die DataFrameWriter-Methode, die nach dem Format der zu verwendenden Dateien benannt ist. Siehe Speichern eines DataFrame in Dateien in einem Stagingbereich.
Beispiel für eine SQL-Anweisung |
Beispiel für Snowpark-Code |
|---|---|
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")
|
Erstellen und Aufrufen von benutzerdefinierten Funktionen (UDFs)¶
Um eine Scala-Funktion zu erstellen, die als UDF (eine anonyme UDF) dient, verwenden Sie udf.
Um eine temporäre oder permanente UDF zu erstellen, die Sie mit Namen aufrufen können, verwenden Sie UDFRegistration.registerTemporary oder UDFRegistration.registerPermanent.
Um eine permanente UDF mit Namen aufzurufen, verwenden Sie callUDF.
Weitere Details dazu finden Sie unter Erstellen von benutzerdefinierten Funktionen (UDFs) für DataFrames in Scala und Aufrufen von skalaren benutzerdefinierten Funktionen (UDFs).
Beispiel für eine SQL-Anweisung |
Beispiel für Snowpark-Code |
|---|---|
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()
|
Erstellen und Aufrufen von gespeicherten Prozeduren¶
Eine Anleitung zum Erstellen von gespeicherten Prozeduren mit Snowpark finden Sie unter Erstellen von gespeicherten Prozeduren für DataFrames in Scala.
Um eine anonyme oder benannte temporäre Prozedur zu erstellen, verwenden Sie eine
registerTemporary-Methode der Klasse com.snowflake.snowpark.SProcRegistration.Um eine benannte permanente Prozedur zu erstellen, verwenden Sie eine
registerPermanent-Methode der Klasse com.snowflake.snowpark.SProcRegistration.Um eine gespeicherte Prozedur aufzurufen, verwenden Sie die Methode
storedProcedureder Klasse com.snowflake.snowpark.Session.
Beispiel für eine SQL-Anweisung |
Beispiel für Snowpark-Code |
|---|---|
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
|