Kurzübersicht: Snowpark-Java-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 select.

Beispiel für eine SQL-Anweisung

Beispiel für Snowpark-Code

 
SELECT id, name FROM sample_product_data;
Copy
 
DataFrame df = session.table("sample_product_data");

DataFrame dfSelectedCols = df.select(Functions.col("id"), Functions.col("name"));

dfSelectedCols.show();
Copy

Umbenennen von Spalten

Um eine Spalte umzubenennen, verwenden Sie as oder alias.

Beispiel für eine SQL-Anweisung

Beispiel für Snowpark-Code

 
SELECT id AS item_id FROM sample_product_data;
Copy
 
DataFrame df = session.table("sample_product_data");

DataFrame dfRenamedCol = df.select(Functions.col("id").as("item_id"));

dfRenamedCol.show();
Copy
 
DataFrame df = session.table("sample_product_data");

DataFrame dfRenamedCol = df.select(Functions.col("id").alias("item_id"));

dfRenamedCol.show();
Copy

Filtern von Daten

Zum Filtern von Daten verwenden Sie filter oder where.

Beispiel für eine SQL-Anweisung

Beispiel für Snowpark-Code

 
SELECT * FROM sample_product_data WHERE id = 1;
Copy
 
DataFrame df = session.table("sample_product_data");

DataFrame dfFilteredRows = df.filter(Functions.col("id").equal_to(Functions.lit(1)));

dfFilteredRows.show();
Copy
 
DataFrame df = session.table("sample_product_data");

DataFrame dfFilteredRows = df.where(Functions.col("id").equal_to(Functions.lit(1)));

dfFilteredRows.show();
Copy

Sortieren von Daten

Zum Sortieren von Daten verwenden Sie sort.

Beispiel für eine SQL-Anweisung

Beispiel für Snowpark-Code

 
SELECT * FROM sample_product_data ORDER BY category_id;
Copy
 
DataFrame df = session.table("sample_product_data");

DataFrame dfSorted = df.sort(Functions.col("category_id"));

dfSorted.show();
Copy

Begrenzen der zurückgegebenen Zeilenanzahl

Um die Anzahl der zurückgegebenen Zeilen zu begrenzen, verwenden Sie 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;
Copy
 
DataFrame df = session.table("sample_product_data");

DataFrame dfSorted = df.sort(Functions.col("category_id")).limit(2);

Row[] arrayRows = dfSorted.collect();
Copy

Ausführen von Verknüpfungen (Joins)

Verwenden Sie zum Verknüpfen join oder 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;
Copy
 
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
 
SELECT * FROM sample_a NATURAL JOIN sample_b;
Copy
 
DataFrame dfLhs = session.table("sample_a");

DataFrame dfRhs = session.table("sample_b");

DataFrame dfJoined = dfLhs.naturalJoin(dfRhs);

dfJoined.show();
Copy

Abfragen von semistrukturierten Daten

Verwenden Sie zum Durchlaufen semistrukturierter Daten subField(„<Feldname>“) und subField(<Index>). Siehe Verwenden von semistrukturierten Daten.

Beispiel für eine SQL-Anweisung

Beispiel für Snowpark-Code

 
SELECT src:salesperson.name FROM car_sales;
Copy
 
DataFrame df = session.table("car_sales");

DataFrame dfJsonField =
  df.select(Functions.col("src").subField("salesperson").subField("name"));

dfJsonField.show();
Copy

Gruppieren und Aggregieren von Daten

Verwenden Sie zum Gruppieren von Daten 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;
Copy
 
DataFrame df = session.table("sample_product_data");

DataFrame dfCountPerCategory = df.groupBy(Functions.col("category_id")).count();

dfCountPerCategory.show();
Copy

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, id, SUM(amount) OVER
  (PARTITION BY category_id ORDER BY product_date)
  FROM sample_product_data ORDER BY product_date;
Copy
 
WindowSpec window = Window.partitionBy(
  Functions.col("category_id")).orderBy(Functions.col("product_date"));

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

DataFrame dfCumulativePrices = df.select(
  Functions.col("category_id"), Functions.col("product_date"),
  Functions.sum(Functions.col("amount")).over(window)).sort(Functions.col("product_date"));

dfCumulativePrices.show();
Copy

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;
Copy
 
import java.util.HashMap;
import java.util.Map;
...

Map<Column, Column> assignments = new HashMap<>();

assignments.put(Functions.col("serial_number"), Functions.lit("xyz"));

Updatable updatableDf = session.table("sample_product_data");

UpdateResult updateResult =
  updatableDf.update(
    assignments,
    Functions.col("id").equal_to(Functions.lit(12)));

System.out.println("Number of rows updated: " + updateResult.getRowsUpdated());
Copy
 
DELETE FROM sample_product_data
  WHERE category_id = 50;
Copy
 
Updatable updatableDf = session.table("sample_product_data");

DeleteResult deleteResult =
  updatableDf.delete(updatableDf.col("category_id").equal_to(Functions.lit(50)));

System.out.println("Number of rows deleted: " + deleteResult.getRowsDeleted());
Copy
 
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;
Copy
 
import java.util.HashMap;
import java.util.Map;

Map<String, Column> assignments = new HashMap<>();
assignments.put("description", source.col("description"));
MergeResult mergeResult =
   target.merge(source, target.col("id").equal_to(source.col("id")))
  .whenMatched.updateColumn(assignments)
  .collect();
Copy

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;
Copy
 
import java.util.HashMap;
import java.util.Map;
...
Map<String, String> putOptions = new HashMap<>();

putOptions.put("OVERWRITE", "TRUE");

PutResult[] putResults = session.file().put(
  "file:///tmp/*.csv", "@myStage", putOptions);

for (PutResult result : putResults) {
  System.out.println(result.getSourceFileName() + ": " + result.getStatus());
}
Copy
 
GET @myStage file:///tmp PATTERN = '.*.csv.gz';
Copy
 
import java.util.HashMap;
import java.util.Map;
...
Map<String, String> getOptions = new HashMap<>();

getOptions.put("PATTERN", "'.*.csv.gz'");

GetResult[] getResults = session.file().get(
 "@myStage", "file:///tmp", getOptions);

for (GetResult result : getResults) {
  System.out.println(result.getFileName() + ": " + result.getStatus());
}
Copy

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;
Copy
 
DataFrame df = session.read().json(
  "@mystage/car_sales.json").select(
    Functions.col("$1").subField(0).subField("salesperson").subField("name"));

df.show();
Copy

Kopieren von Daten aus Dateien in einem Stagingbereich in eine Tabelle

Verwenden Sie zum Kopieren von Dateien aus einem Stagingbereich in eine Tabelle die DataFrameReader-Klasse, mit der Sie ein CopyableDataFrame-Objekt für die Daten erstellen, und verwenden Sie die Methode 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);
Copy
 
CopyableDataFrame dfCopyableDf = session.read().json("@mystage/car_sales.json");
dfCopyableDf.copyInto("new_car_sales");
Copy

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
Copy
 
DataFrame df = session.table("car_sales");

WriteFileResult writeFileResult = df.write().mode(
  SaveMode.Overwrite).option(
  "DETAILED_OUTPUT", "TRUE").option(
  "compression", "none").json(
  "@mystage/saved_data.json");
Copy

Erstellen und Aufrufen von benutzerdefinierten Funktionen (UDFs)

Um einen anonymen UDF zu erstellen, verwenden Sie Functions.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 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>(quantity) AS doublenum
  FROM sample_product_data;
Copy
 
UserDefinedFunction doubleUdf =
  Functions.udf(
    (Integer x) -> x + x,
    DataTypes.IntegerType,
    DataTypes.IntegerType);

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

DataFrame dfWithDoubleNum =
  df.withColumn("doubleNum",
    doubleUdf.apply(Functions.col("quantity")));

dfWithDoubleNum.show();
Copy
 
CREATE FUNCTION <temp_function_name>
  RETURNS INT
  LANGUAGE JAVA
  ...
  AS
  ...;

SELECT ...,
  <temp_function_name>(quantity) AS doublenum
  FROM sample_product_data;
Copy
 
UserDefinedFunction doubleUdf =
  session
    .udf()
    .registerTemporary(
      "doubleUdf",
      (Integer x) -> x + x,
      DataTypes.IntegerType,
      DataTypes.IntegerType);

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

DataFrame dfWithDoubleNum =
  df.withColumn("doubleNum",
    Functions.callUDF("doubleUdf", Functions.col("quantity")));
dfWithDoubleNum.show();
Copy
 
CREATE FUNCTION doubleUdf(arg1 INT)
  RETURNS INT
  LANGUAGE JAVA
  ...
  AS
  ...;

SELECT ...,
  doubleUdf(quantity) AS doublenum
  FROM sample_product_data;
Copy
 
UserDefinedFunction doubleUdf =
  session
    .udf()
    .registerPermanent(
      "doubleUdf",
      (Integer x) -> x + x,
      DataTypes.IntegerType,
      DataTypes.IntegerType,
      "mystage");

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

DataFrame dfWithDoubleNum =
  df.withColumn("doubleNum",
    Functions.callUDF("doubleUdf", Functions.col("quantity")));
dfWithDoubleNum.show();
Copy

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

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);
Copy
 
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();
Copy
 
CREATE PROCEDURE sproc(x INTEGER, y INTEGER)
  RETURNS INTEGER
  LANGUAGE JAVA
  ...
  AS
  $$
  BEGIN
   RETURN x + y;
  END
  $$
  ;

CALL sproc(2, 3);
Copy
 
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();
Copy
 
CREATE PROCEDURE add_hundred(x INTEGER)
  RETURNS INTEGER
  LANGUAGE JAVA
  ...
  AS
  $$
  BEGIN
   RETURN x + 100;
  END
  $$
  ;

CALL add_hundred(3);
Copy
 
String name = "add_hundred";
String stageName = "sproc_libs";

StoredProcedure sp =
  session.sproc().registerPermanent(
    name,
    (Session session, Integer x) -> x + 100,
    DataTypes.IntegerType,
    DataTypes.IntegerType,
    stageName,
    true);

  session.storedProcedure(name, 3).show();
Copy
Sprache: Deutsch