Verwenden von DataFrames in Snowpark Java¶

Angeben von Spalten und Ausdrücken¶

Wenn Sie diese Transformationsmethoden aufrufen, müssen Sie möglicherweise Spalten angeben oder Ausdrücke, die Spalten verwenden. Wenn Sie z. B. die select-Methode aufrufen, müssen Sie die Spalten angeben, die ausgewählt werden sollen.

Um auf eine Spalte zu verweisen, erstellen Sie ein Column-Objekt durch Aufruf der statischen Methode Functions.col.

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

Wenn Sie einen Filter, eine Projektion, eine Verknüpfungsbedingung usw. angeben müssen, können Sie Column-Objekte in einem Ausdruck verwenden. Beispiel:

• Sie können Column-Objekte mit der Methode filter verwenden, um eine Filterbedingung anzugeben:

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

• Sie können Column-Objekte mit der Methode select verwenden, um einen Alias zu definieren:

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

• Sie können Column-Objekte mit der Methode join verwenden, um eine Verknüpfungsbedingung zu definieren:

  // 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();

Verwenden von doppelten Anführungszeichen um Objektbezeichner (Tabellennamen, Spaltennamen usw.)¶

Die Namen von Datenbanken, Schemas, Tabellen und Stagingbereichen, die Sie angeben, müssen den Snowflake-Anforderungen an Bezeichner entsprechen. Wenn Sie einen Namen angeben, geht Snowflake davon aus, dass der Name in Großbuchstaben geschrieben ist. Daher sind z. B. die folgenden Aufrufe gleichwertig:

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

Wenn der Name nicht den Anforderungen für Bezeichner entspricht, müssen Sie den Namen in doppelte Anführungszeichen (") setzen. Verwenden Sie einen Backslash (\) als Escape-Zeichen zum Umschließen von doppelten Anführungszeichen innerhalb von Scala-Zeichenfolgenliteralen. Der folgende Tabellenname beginnt z. B. nicht mit einem Buchstaben oder einem Unterstrich, sodass Sie den Namen in doppelte Anführungszeichen setzen müssen:

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

Beachten Sie, dass Sie bei der Angabe des Namens einer Spalte keine doppelten Anführungszeichen um den Namen verwenden müssen. Die Snowpark-Bibliothek schließt Spaltennamen automatisch in doppelte Anführungszeichen ein, wenn der Name nicht den Bezeichneranforderungen entspricht:

// 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\""));

Wenn Sie bereits doppelte Anführungszeichen um einen Spaltennamen hinzugefügt haben, fügt die Bibliothek keine weiteren doppelten Anführungszeichen um den Namen ein.

In einigen Fällen kann der Spaltenname doppelte Anführungszeichen enthalten:

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

Wie unter Anforderungen an Bezeichner erläutert, müssen Sie für jedes doppelte Anführungszeichen innerhalb eines in Anführungszeichen geschriebenen Bezeichners zwei doppelte Anführungszeichen verwenden (z. B. "name_with_""air""_quotes" und """column_name_quoted"""):

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

Wenn ein Bezeichner in doppelte Anführungszeichen eingeschlossen ist (unabhängig davon, ob Sie die Anführungszeichen explizit hinzugefügt haben oder die Bibliothek die Anführungszeichen für Sie hinzugefügt hat), berücksichtigt Snowflake die Groß-/Kleinschreibung des Bezeichners:

// 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"));

Verwenden von Literalen als Spaltenobjekte¶

Um ein Literal in einer Methode zu verwenden, die ein Column-Objekt übergibt, erstellen Sie ein Column-Objekt für das Literal, indem Sie das Literal an die statische Methode lit in der Functions-Klasse übergeben. Beispiel:

// 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();

Wenn das Literal in Java ein Gleitkomma- oder Double-Wert ist (z. B. 0.05 wird standardmäßig als Double verarbeitet), generiert die Snowpark-Bibliothek SQL-Code, der den Wert implizit in den entsprechenden Snowpark-Datentyp umwandelt (z. B. 0.05::DOUBLE). Dies kann zu einem Näherungswert führen, der von der genau angegebenen Zahl abweicht.

Der Code im folgenden Beispiel zeigt keine übereinstimmenden Zeilen an, obwohl der Filter (der für Werte größer oder gleich 0.05 gilt) mit den Zeilen in DataFrame übereinstimmen sollte:

// 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();

Das Problem ist, dass Functions.lit(0.06) und Functions.lit(0.01) Näherungswerte für 0.06 und 0.01 liefern und nicht die genauen Werte.

Zur Umgehung dieses Problems wandeln Sie das Literal in den Snowpark-Typ um, den Sie verwenden möchten. Verwenden Sie zum Beispiel einen NUMBER mit einer Gesamtstellenzahl (precision) von 5 und einer Dezimalstellenzahl (scale) von 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();

Umwandeln eines Spaltenobjekts in einen bestimmten Typ¶

Um ein Column-Objekt in einen bestimmten Typ umzuwandeln, rufen Sie die cast-Methode auf und übergeben ein Typobjekt aus dem com.snowflake.snowpark_java.types-Paket. So können Sie zum Beispiel ein Literal als NUMBER mit einer Gesamtstellenzahl (precision) von 5 und einer Dezimalstellenzahl (scale) von 2 darstellen:

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

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

Verketten von Methodenaufrufen¶

Da jede Methode, die ein DataFrame-Objekt transformiert, ein neues DataFrame-Objekt zurückgibt, auf das die Transformation angewendet wurde, können Sie Methodenaufrufe verketten, um ein neues DataFrame zu erstellen, das auf eine andere Weise transformiert wird.

Im folgenden Beispiel wird ein DataFrame zurückgegeben, der für folgende Aufgaben konfiguriert ist:

• Abfragen der Tabelle sample_product_data

• Rückgabe der Zeile mit id = 1

• Auswahl der Spalten name und 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();

In diesem Beispiel:

• session.table("sample_product_data") gibt einen DataFrame für die Tabelle sample_product_data zurück.

  Der DataFrame enthält zwar noch nicht die Daten aus der Tabelle, aber das Objekt enthält die Definitionen der Spalten in der Tabelle.

• filter(Functions.col("id").equal_to(Functions.lit(1))) gibt ein DataFrame für die Tabelle sample_product_data zurück, die so eingerichtet ist, dass sie die Zeile mit id = 1 zurückgibt.

  Beachten Sie erneut, dass das DataFrame noch nicht die übereinstimmende Zeile aus der Tabelle enthält. Die übereinstimmende Zeile wird erst abgerufen, wenn Sie eine Aktionsmethode aufrufen.

• select(Functions.col("name"), Functions.col("serial_number")) gibt ein DataFrame zurück, das die Spalten name und serial_number für die Zeile in der Tabelle sample_product_data enthält, die id = 1 hat.

Beachten Sie beim Verketten von Methodenaufrufen, dass die Reihenfolge der Aufrufe wichtig ist. Jeder Methodenaufruf gibt ein DataFrame zurück, das transformiert wurde. Stellen Sie sicher, dass nachfolgende Aufrufe das transformierte DataFrame verwenden.

Im folgenden Beispiel gibt die Methode select ein DataFrame zurück, das nur zwei Spalten enthält: name und serial_number. Der Aufruf der filter-Methode auf diesem DataFrame schlägt fehl, weil dieser die Spalte id verwendet, die im transformierten DataFrame nicht vorhanden ist.

// 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();

Im Gegensatz dazu wird der folgende Code erfolgreich ausgeführt, weil die filter()-Methode auf einem DataFrame aufgerufen wird, der alle Spalten der Tabelle sample_product_data enthält (einschließlich der Spalte 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();

Beachten Sie, dass Sie die Methodenaufrufe select und filter möglicherweise in einer anderen Reihenfolge ausführen müssen als bei Verwendung der entsprechenden Schlüsselwörter (SELECT und WHERE) in einer SQL-Anweisung.

Begrenzen der Anzahl von Zeilen in einem DataFrame¶

Um die Anzahl der Zeilen in einem DataFrame zu begrenzen, können Sie die Transformationsmethode limit verwenden.

Die Snowpark-API bietet auch folgende Aktionsmethoden zum Abrufen und Ausdrucken einer begrenzten Anzahl von Zeilen:

• Aktionsmethode first (zum Ausführen von Abfragen und zum Zurückgeben der ersten n Zeilen)

• Aktionsmethode show (zum Ausführen von Abfragen und zum Drucken der ersten n Zeilen)

Diese Methoden fügen der auszuführenden SQL-Anweisung im Endeffekt eine LIMIT-Klausel hinzu.

Wie in den Nutzungshinweisen zu LIMIT erläutert, sind die Ergebnisse nicht deterministisch, es sei denn, Sie geben in Verbindung mit LIMIT eine Sortierreihenfolge (ORDER BY) an.

Um die ORDER BY-Klausel zusammen mit der LIMIT-Klausel zu verwenden (z. B. damit ORDER BY nicht in einer separaten Unterabfrage steht), müssen Sie die Methode aufrufen, die die Ergebnisse auf den von der sort-Methode zurückgegebenen DataFrame beschränkt.

Wenn Sie beispielsweise Methodenaufrufe verketten:

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);

Abrufen von Spaltendefinitionen¶

Um die Definition der Spalten im Dataset für DataFrame abzurufen, verwenden Sie die schema-Methode. Diese Methode gibt ein StructType-Objekt zurück, das ein Array von StructField-Objekten enthält. Jedes StructField-Objekt enthält die Definition einer Spalte.

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);

Im zurückgegebenen StructType-Objekt sind die Spaltennamen immer normalisiert. Bezeichner, die nicht mit Anführungszeichen umschlossen sind, werden in Großbuchstaben zurückgegeben. Bezeichner in Anführungszeichen werden genau in der Schreibweise zurückgegeben, in der sie definiert wurden.

Im folgenden Beispiel wird ein DataFrame erstellt zurück, der die Spalten mit den Namen ID und 3rd enthält. Für den Spaltennamen 3rd schließt die Snowpark-Bibliothek den Namen automatisch in doppelte Anführungszeichen ein ("3rd"), da der Name nicht den Anforderungen an einen Bezeichner entspricht.

Im Beispiel wird erst die schema-Methode aufgerufen und dann die names-Methode auf dem zurückgegebenen Objekt StructType, um ein Array von Spaltennamen zu erhalten. Die Namen werden in dem von der schema-Methode zurückgegebenen StructType normalisiert.

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()));

Verknüpfen von DataFrames¶

Zum Verknüpfen (Join) von DataFrame-Objekten rufen Sie die join-Methode auf:

In den folgenden Abschnitten wird erläutert, wie DataFrames zur Durchführung einer Verknüpfung verwendet werden:

• Einrichten der Beispieldaten für die Verknüpfungen

• Angeben der Spalten für den Join

• Ausführen einer natürlichen Verknüpfung (Natural Join)

• Geben Sie den Typ der Verknüpfung an:

• Mehrere Tabellen verknüpfen

• Ausführen einer Selbstverknüpfung (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".
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");

Synchrones Ausführen einer Aktion¶

Um eine Aktion synchron auszuführen, rufen Sie eine der folgenden Aktionsmethoden auf:

Um beispielsweise die Abfrage auszuführen und die Anzahl der Ergebnisse zurückzugeben, rufen Sie die count-Methode auf:

// 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());

Sie können auch Aktionsmethoden aufrufen, um Folgendes zu tun:

• Ausführen einer Abfrage auf einer Tabelle und Zurückgeben der Ergebnisse

• Ausführen einer Abfrage und Ausgeben der Ergebnisse auf einer Konsole

Hinweis: Wenn Sie die schema-Methode aufrufen, um die Definitionen der Spalten im DataFrame zu erhalten, müssen Sie keine Aktionsmethode aufrufen.

Asynchrones Ausführen einer Aktion¶

Zur asynchronen Ausführung einer Aktion müssen Sie die Methode async aufrufen, um ein Objekt für den „asynchronen Akteur“ (z. B. DataFrameAsyncActor) zurückzugeben, und dann eine asynchrone Aktionsmethode in diesem Objekt aufrufen.

Diese Aktionsmethoden eines asynchronen Akteursobjekts geben ein TypedAsyncJob-Objekt zurück, das Sie verwenden können, um den Status der asynchronen Aktion zu überprüfen und die Ergebnisse der Aktion abzurufen.

In den nächsten Abschnitten wird erläutert, wie Sie Aktionen asynchron ausführen und die Ergebnisse überprüfen können.

• Erläuterungen zum grundlegenden Ablauf asynchroner Aktionen

• Festlegen der maximalen Anzahl der zu wartenden Sekunden

• Zugriff auf eine asynchrone Abfrage über die 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));

Alle Zeilen zurückgeben¶

Um alle Zeilen auf einmal zurückzugeben, rufen Sie die Methode collect auf. Diese Methode gibt ein Array von Row-Objekten zurück. Um die Werte aus der Zeile abzurufen, rufen Sie die Methode getType auf, wobei „Type“ der Datentyp ist (z. B. getString, getInt usw.).

Beispiel:

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));
}

Zurückgeben eines Iterators für die Zeilen¶

Wenn Sie einen Iterator verwenden möchten, um die Row-Objekte in den Ergebnissen zeilenweise zu durchlaufen, rufen Sie toLocalIterator auf. Wenn die Datenmenge in den Ergebnissen groß ist, lädt die Methode die Zeilen blockweise, um zu vermeiden, dass alle Zeilen auf einmal in den Arbeitsspeicher geladen werden.

Beispiel:

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));
}

Zurückgeben der ersten n Zeilen¶

Um die ersten n Zeilen zurückzugeben, rufen Sie die Methode first auf und übergeben die Anzahl der zurückzugebenden Zeilen.

Wie unter Begrenzen der Anzahl von Zeilen in einem DataFrame erläutert, sind die Ergebnisse nicht deterministisch. Wenn die Ergebnisse deterministisch sein sollen, müssen Sie diese Methode auf einem sortierten DataFrame (df.sort().first()) aufrufen.

Beispiel:

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));

Aktualisieren von Zeilen in einer Tabelle¶

Um die Zeilen einer Tabelle zu aktualisieren, rufen Sie die Methode update oder updateColumn auf und übergeben eine Map, die die Zuordnung der zu aktualisierenden Spalten und der entsprechenden Werte, die diesen Spalten zugewiesen werden sollen, enthält:

• Um die Spaltennamen als Zeichenfolgen in der Map anzugeben, rufen Sie updateColumn auf.

• Um Column-Objekte in Map anzugeben, rufen Sie update auf.

Beide Methoden geben ein UpdateResult-Objekt zurück, das die Anzahl der aktualisierten Zeilen enthält. (siehe UpdateResult).

Wenn Sie beispielsweise die Werte in der Spalte mit dem Namen count durch den Wert 1 ersetzen und dabei eine Map verwenden möchten, die den Spaltennamen (ein String-Wert) mit dem entsprechenden Wert verknüpft, rufen Sie updateColumn auf:

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());

Wenn Sie in der ein Map ein Column-Objekt verwenden möchten, um die zu aktualisierende Spalte zu identifizieren, rufen Sie update auf:

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());

Wenn die Aktualisierung nur erfolgen soll, wenn eine Bedingung erfüllt ist, können Sie diese Bedingung als Argument angeben. Im folgenden Beispiel wird der Wert in der Spalte mit dem Namen count durch 2 für die Zeilen ersetzt, in denen die Spalte category_id den Wert 20 hat:

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());

Wenn die Bedingung auf einer Verknüpfung mit einem anderen DataFrame-Objekt basieren soll, können Sie dieses DataFrame als Argument übergeben und dieses DataFrame in der Bedingung verwenden. Im folgenden Beispiel wird der Wert in der Spalte mit dem Namen count durch 3 für die Zeilen ersetzt, in denen die Spalte category_id mit der category_id in der DataFrame-Spalte dfParts übereinstimmt:

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());

Löschen von Zeilen aus einer Tabelle¶

Sie können für die Methode delete eine Bedingung angeben, die die zu löschenden Zeilen identifiziert, und diese Bedingung kann auf einer Verknüpfung (Join) mit einem anderen DataFrame basieren. delete gibt ein DeleteResult-Objekt zurück, das die Anzahl der gelöschten Zeilen enthält (siehe DeleteResult).

Im folgenden Beispiel werden Zeilen gelöscht, die den Wert 1 in der Spalte category_id haben:

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());

Wenn sich die Bedingung auf Spalten in einem anderen DataFrame bezieht, übergeben Sie dieses DataFrame als zweites Argument. Im folgenden Beispiel werden die Zeilen gelöscht, in denen die Spalte category_id mit der category_id in der DataFrame-Spalte dfParts übereinstimmt, wobei dfParts als zweites Argument übergeben wird:

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());

Zusammenführen (Merge) von Zeilen in einer Tabelle¶

Wenn Sie Zeilen einer Tabelle einfügen, aktualisieren oder löschen möchten, die auf Werten einer zweiten Tabelle oder einer Unterabfrage basieren (das Äquivalent des MERGE-Befehls in SQL), gehen Sie wie folgt vor:

1. Rufen Sie im Updatable-Objekt der Tabelle, in der die Daten zusammengeführt werden sollen, die Methode merge auf, und übergeben Sie dabei das DataFrame-Objekt für die andere Tabelle und den Spaltenausdruck für die Verknüpfungsbedingung.

   Dies gibt ein MergeBuilder-Objekt zurück, das Sie verwenden können, um die Aktionen (z. B. Einfügen, Aktualisieren, Löschen) für die übereinstimmenden und die nicht übereinstimmenden Zeilen anzugeben (siehe MergeBuilder).

2. Verwendung des Objekts MergeBuilder:

   • Um die Aktualisierung oder Löschung von übereinstimmenden Zeilen anzugeben, rufen Sie die Methode whenMatched auf.

     Wenn Sie eine zusätzliche Bedingung angeben müssen, ob Zeilen aktualisiert oder gelöscht werden sollen, können Sie einen Spaltenausdruck für diese Bedingung übergeben.

     Diese Methode gibt ein MatchedClauseBuilder-Objekt zurück, das Sie zur Angabe der auszuführenden Aktion verwenden können (siehe MatchedClauseBuilder).

     Rufen Sie die Methode update oder delete im Objekt MatchedClauseBuilder auf, um die Aktualisierungs- oder Löschaktion anzugeben, die für die übereinstimmenden Zeilen durchgeführt werden soll. Diese Methoden geben ein MergeBuilder-Objekt zurück, das Sie zur Angabe zusätzlicher Klauseln verwenden können.

   • Um die Einfügung anzugeben, die ausgeführt werden soll, wenn Zeilen nicht übereinstimmen, rufen Sie die Methode whenNotMatched auf.

     Wenn Sie für das Einfügen einer Zeile eine zusätzliche Bedingung angeben müssen, können Sie einen Spaltenausdruck für diese Bedingung übergeben.

     Diese Methode gibt ein NotMatchedClauseBuilder-Objekt zurück, das Sie zur Angabe der auszuführenden Aktion verwenden können (siehe NotMatchedClauseBuilder).

     Rufen Sie die Methode insert im Objekt NotMatchedClauseBuilder auf, um die Einfügeaktion festzulegen, die durchgeführt werden soll, wenn Zeilen nicht übereinstimmen. Diese Methoden geben ein MergeBuilder-Objekt zurück, das Sie zur Angabe zusätzlicher Klauseln verwenden können.

3. Wenn Sie die auszuführenden Einfügungen, Aktualisierungen und Löschungen angegeben haben, rufen Sie die Methode collect des Objekts MergeBuilder auf, um die angegebenen Einfügungen, Aktualisierungen und Löschungen in der Tabelle auszuführen.

   collect gibt ein MergeResult-Objekt zurück, das die Anzahl der Zeilen enthält, die eingefügt, aktualisiert und gelöscht wurden (siehe MergeResult).

Das folgende Beispiel fügt eine Zeile mit den Spalten id und value aus der Tabelle source in die Tabelle target ein, wenn die Tabelle target keine Zeile mit einer übereinstimmenden ID enthält:

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

Im folgenden Beispiel wird eine Zeile in der Tabelle target mit dem Wert der Spalte value aus der Zeile in der Tabelle source aktualisiert, die dieselbe ID hat:

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();

Hochladen und Herunterladen von Dateien in Stagingbereichen¶

Zum Hoch- und Herunterladen von Dateien eines Stagingbereichs verwenden Sie die Methoden put und get des Objekts FileOperation:

• Hochladen von Dateien in einen Stagingbereich

• Herunterladen von Dateien aus Stagingbereichen

Verwenden von Eingabestreams zum Hoch- und Herunterladen von Daten eines Stagingbereichs¶

Um Eingabestreams zum Hochladen von Daten in eine Stagingdatei und zum Herunterladen von Daten aus einer Stagingdatei zu verwenden, verwenden Sie die Methoden uploadStream und downloadStream des Objekts FileOperation:

• Verwenden eines Eingabestreams zum Hochladen von Daten in eine Datei in einem Stagingbereich

• Verwenden eines Eingabestreams zum Herunterladen von Daten aus einer Datei in einem Stagingbereich

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);

Einrichten eines DataFrame für Dateien in einem Stagingbereich¶

In diesem Abschnitt wird erläutert, wie Sie ein DataFrame-Objekt für Dateien einrichten, die sich in einem Snowflake-Stagingbereich befindet. Sobald Sie den DataFrame erstellt haben, können Sie den DataFrame für Folgendes verwenden:

• Abrufen von Daten aus den Dateien

• Kopieren von Daten aus den Dateien in eine Tabelle

Um einen DataFrame für eine Datei einzurichten, die sich ein einem Snowflake-Stagingbereich befindet, verwenden Sie die Klasse DataFrameReader:

1. Überprüfen Sie, ob Sie über die folgenden Berechtigungen verfügen:

   • Berechtigungen für den Zugriff auf Dateien in dem Stagingbereich.

   • Eine der folgenden Optionen:

     • CREATE TABLE-Berechtigungen für das Schema, wenn Sie Kopieroptionen angeben möchten, die bestimmen, wie die Daten aus den Stagingdateien kopiert werden.

     • Andernfalls CREATE FILE FORMAT-Berechtigungen für das Schema.

2. Rufen Sie die Methode read der Session-Klasse auf, um auf ein DataFrameReader-Objekt zuzugreifen.

3. Wenn die Dateien im CSV-Format sind, beschreiben Sie die Felder in der Datei. Gehen Sie dabei wie folgt vor:

   1. Erstellen Sie ein StructType-Objekt, das aus einem Array von StructField-Objekten besteht, die die Felder in der Datei beschreiben.

   2. Geben Sie für jedes StructField-Objekt Folgendes an:

      • Name des Feldes

      • Datentyp des Feldes (angegeben als Objekt im com.snowflake.snowpark_java.types-Paket)

      • Ob das Feld nullwertfähig ist oder nicht

      Beispiel:

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

      Copy

   3. Rufen Sie die Methode schema im Objekt DataFrameReader-Objekt auf, und übergeben Sie dabei das StructType-Objekt.

      Beispiel:

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

      Copy

      Die Methode schema gibt ein DataFrameReader-Objekt zurück, das so konfiguriert ist, dass es Dateien liest, die die angegebenen Felder enthalten.

      Beachten Sie, dass Sie dies für Dateien in anderen Formaten (z. B. JSON) nicht tun müssen. Für solche Dateien behandelt der DataFrameReader die Daten als ein einzelnes Feld vom Typ VARIANT mit dem Feldnamen $1.

4. Wenn Sie zusätzliche Informationen darüber angeben müssen, wie die Daten gelesen werden sollen (z. B. dass die Daten komprimiert sind oder dass eine CSV-Datei ein Semikolon statt eines Kommas zum Begrenzen von Feldern verwendet), rufen Sie die Methode DataFrameReader.option oder DataFrameReader.options auf.

   Geben Sie den Namen und den Wert der Option ein, die Sie einstellen möchten. Sie können die folgenden Typen von Optionen einstellen:

   • Dateiformatoptionen, die in der CREATE FILE FORMAT-Dokumentation beschrieben sind

   • Kopieroptionen, die in der COPY INTO TABLE-Dokumentation beschrieben sind

     Beachten Sie, dass die Einstellung von Kopieroptionen zu einer teureren Ausführungsstrategie führen kann, wenn Sie die Daten in den DataFrame abrufen.

   Im folgenden Beispiel wird das DataFrameReader-Objekt für die Abfrage von Daten einer CSV-Datei eingerichtet, die nicht komprimiert ist und die ein Semikolon als Feldbegrenzer verwendet.

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

   Copy

   Die Methode option gibt ein DataFrameReader-Objekt zurück, das mit den angegebenen Optionen konfiguriert ist.

   Um mehrere Optionen festzulegen, können Sie entweder verkettete Aufrufe mit der option-Methode verwenden (wie im obigen Beispiel gezeigt) oder die DataFrameReader.options-Methode aufrufen, wobei Sie ein Map-Objekt der Namen und Werte der Optionen übergeben.

5. Rufen Sie die Methode auf, die dem Format der Dateien entspricht: Sie können eine der folgenden Methoden aufrufen:

   • DataFrameReader.avro

   • DataFrameReader.csv

   • DataFrameReader.json

   • DataFrameReader.orc

   • DataFrameReader.parquet

   • DataFrameReader.xml

   Beim Aufrufen dieser Methoden übergeben Sie den Speicherort des Stagingbereichs der zu lesenden Dateien. Beispiel:

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

   Copy

   Wenn Sie mehrere Dateien angeben möchten, die mit demselben Präfix beginnen, geben Sie das Präfix nach dem Namen des Stagingbereichs an. So laden Sie beispielsweise Dateien mit dem Präfix csv_ aus dem Stagingbereich @mystage:

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

   Copy

   Die Methoden, die dem Format einer Datei entsprechen, geben ein CopyableDataFrame-Objekt für diese Datei zurück. CopyableDataFrame erweitert DataFrame und bietet zusätzliche Methoden für die Verarbeitung der Daten in den Stagingdateien.

6. Rufen Sie eine Aktionsmethode auf, um Folgendes auszuführen:

   • Abrufen von Daten aus Dateien oder

   • Kopieren von Daten aus den Dateien in eine Tabelle

   Wie bei DataFrames für Tabellen werden die Daten erst dann in den DataFrame abgerufen, wenn Sie eine Aktionsmethode aufrufen.

Laden von Daten aus Dateien in einen DataFrame¶

Nachdem Sie einen DataFrame für Dateien in einem Stagingbereich eingerichtet haben, können Sie Daten aus den Dateien in den DataFrame laden:

1. Verwenden Sie die DataFrame-Objektmethoden, um alle Transformationen auszuführen, die für das Dataset benötigt werden (Auswahl bestimmter Felder, Filtern von Zeilen usw.).

   Im folgenden Beispiel wird das Element color aus einer JSON-Datei namens data.json extrahiert, die sich im Stagingbereich mystage befindet:

   DataFrame df = session.read().json("@mystage/data.json").select(Functions.col("$1").subField("color"));
   

   Copy

   Wie bereits erläutert, werden Daten in Dateien, die nicht im CSV-Format sind (z. B. JSON), von DataFrameReader als eine einzelne VARIANT-Spalte mit dem Namen $1 behandelt.

2. Rufen Sie die Methode DataFrame.collect auf, um die Daten zu laden. Beispiel:

   Row[] results = df.collect();
   

   Copy

Kopieren von Daten aus Dateien in eine Tabelle¶

Nachdem das Einrichten eines DataFrame für Dateien in einem Stagingbereich abgeschlossen ist, können Sie die Methode copyInto aufrufen, um die Daten in eine Tabelle zu kopieren. Diese Methode führt den Befehl COPY INTO <Tabelle> aus.

Mit dem folgenden Code werden beispielsweise Daten aus der durch myFileStage angegebenen CSV-Datei in die Tabelle mytable geladen. Da sich die Daten in einer CSV-Datei befinden, muss der Code auch die Felder in der Datei beschreiben. Im Beispiel wird dazu die Methode schema des DataFrameReader-Objekts aufgerufen, und es wird ein StructType-Objekt (schemaForDataFile) mit einem Array von StructField-Objekten übergeben, die die Felder beschreiben.

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

Speichern eines DataFrame in Dateien in einem Stagingbereich¶

Wenn Sie einen DataFrame in Dateien in einem Stagingbereich speichern müssen, können Sie die DataFrameWriter-Methode aufrufen, die dem Format der Datei entspricht (z. B. die csv-Methode, um in eine CSV-Datei zu schreiben), und dabei den Speichertort im Stagingbereich angeben, an dem die Dateien gespeichert werden sollen. Diese DataFrameWriter-Methoden führen den Befehl COPY INTO <Speicherort> aus.

So speichern Sie den Inhalt eines DataFrame in Dateien in einem Stagingbereich:

1. Rufen Sie die write-Methode des DataFrame auf, um ein DataFrameWriter-Objekt zu erhalten. So erhalten Sie beispielsweise das DataFrameWriter-Objekt für einen DataFrame, der die Tabelle mit dem Namen sample_product_data repräsentiert:

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

   Copy

2. Wenn Sie den Inhalt der Datei überschreiben möchten (falls die Datei existiert), rufen Sie die Methode mode des Objekts DataFrameWriter auf und übergeben SaveMode.Overwrite.

   Andernfalls meldet der DataFrameWriter-Objekt standardmäßig einen Fehler, wenn die angegebene Datei im Stagingbereich bereits vorhanden ist.

   Diese mode-Methode gibt das gleiche DataFrameWriter-Objekt zurück, das mit dem angegebenen Modus konfiguriert wurde.

   So geben Sie beispielsweise an, dass DataFrameWriter die im Stagingbereich befindliche Datei überschreiben soll:

   dfWriter = dfWriter.mode(SaveMode.Overwrite);
   

   Copy

3. Wenn Sie zusätzliche Informationen darüber angeben müssen, wie die Daten gespeichert werden sollen (z. B. dass die Daten komprimiert werden sollen oder dass Sie ein Semikolon zur Abgrenzung von Feldern in einer CSV-Datei verwenden möchten), dann rufen Sie die Methode DataFrameWriter.option oder DataFrameWriter.options auf.

   Geben Sie den Namen und den Wert der Option ein, die Sie einstellen möchten. Sie können die folgenden Typen von Optionen einstellen:

   • Dateiformatoptionen, die in der Dokumentation zu COPY INTO <Speicherort> beschrieben sind.

   • Kopieroptionen, die in der Dokumentation zu COPY INTO <Speicherort> beschrieben sind.

   • PARTITION BY oder HEADER.

   Beachten Sie, dass Sie die Methode option nicht zum Einstellen der folgenden Optionen verwenden können:

   • Die Option für den Formattyp TYPE.

   • Die Kopieroption OVERWRITE. Um diese Option zu setzen, rufen Sie stattdessen die Methode mode auf (wie im vorherigen Schritt erwähnt).

   Im folgenden Beispiel wird das DataFrameWriter-Objekt so eingerichtet, dass es Daten in einer CSV-Datei in unkomprimierter Form speichert, wobei ein Semikolon (statt eines Kommas) als Trennzeichen verwendet wird.

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

   Copy

   Die Methode option gibt ein DataFrameWriter-Objekt zurück, das mit den angegebenen Optionen konfiguriert ist.

   Um mehrere Optionen zu setzen, können Sie verkettete Aufrufe mit der Methode option verwenden (wie im obigen Beispiel gezeigt) oder die Methode DataFrameWriter.options aufrufen und dabei ein Map-Objekt der Namen und Werte der Optionen übergeben.

4. Um Details über jede gespeicherte Datei zu erhalten, setzen Sie die Kopieroption DETAILED_OUTPUT auf TRUE.

   Standardmäßig hat DETAILED_OUTPUT den Wert FALSE, was bedeutet, dass die Methode eine einzelne Ausgabezeile mit den Feldern "rows_unloaded", "input_bytes" und "output_bytes" zurückgibt.

   Wenn Sie DETAILED_OUTPUT auf TRUE setzen, gibt die Methode für jede gespeicherte Datei eine Zeile der Ausgabe zurück. Jede Zeile enthält die Felder FILE_NAME, FILE_SIZE und ROW_COUNT.

5. Rufen Sie die Methode auf, die dem Format der Datei entspricht, um die Daten in der Datei zu speichern. Sie können eine der folgenden Methoden aufrufen:

   • DataFrameWriter.csv

   • DataFrameWriter.json

   • DataFrameWriter.parquet

   Beim Aufrufen dieser Methoden übergeben Sie den Speicherort des Stagingbereichs der Datei, in die die Daten geschrieben werden sollen (z. B. @mystage).

   Standardmäßig speichert die Methode die Daten in Dateinamen mit dem Präfix data_ (z. B. @mystage/data_0_0_0.csv). Wenn Sie möchten, dass die Dateien mit einem anderen Präfix benannt werden, geben Sie das Präfix nach dem Stagingbereichsnamen an. Beispiel:

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

   Copy

   In diesem Beispiel wird der Inhalt des DataFrame-Objekts in Dateien gespeichert, die mit dem Präfix saved_data beginnen (z. B. @mystage/saved_data_0_0_0.csv).

6. Überprüfen Sie das zurückgegebene WriteFileResult-Objekt auf Informationen zu der in die Datei geschriebene Datenmenge.

   Über das WriteFileResult-Objekt können Sie auf die vom Befehl COPY INTO <Speicherort> erzeugte Ausgabe zugreifen:

   • Um auf die Zeilen der Ausgabe als Array von Row-Objekten zuzugreifen, rufen Sie die Methode getRows auf.

   • Um festzustellen, welche Felder in den Zeilen vorhanden sind, rufen Sie die Methode getSchema auf, die ein StructType-Objekt zurückgibt, das die Felder in der Zeile beschreibt.

   So drucken Sie beispielsweise die Namen der Felder und Werte in den Ausgabezeilen aus:

   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

Im folgende Beispiel wird ein DataFrame-Objekt verwendet, um den Inhalt der Tabelle mit dem Namen car_sales in JSON-Dateien mit dem Präfix saved_data im Stagingbereich @mystage (z. B. @mystage/saved_data_0_0_0.json) zu speichern. Hier der Beispielcode:

• Überschreibt die Datei, wenn die Datei bereits im Stagingbereich vorhanden ist.

• Gibt eine detaillierte Ausgabe zur Speicheroperation zurück.

• Speichert die Daten unkomprimiert.

Abschließend druckt der Beispielcode jedes Feld und jeden Wert in den zurückgegebenen Ausgabezeilen aus:

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));
  }
}

Durchsuchen semistrukturierter Daten¶

Um auf ein bestimmtes Feld oder Element in semistrukturierten Daten zu verweisen, verwenden Sie die folgenden Methoden des Column-Objekts:

• Verwenden Sie subField(„<Feldname>“), um ein Column-Objekt für ein Feld in einem OBJECT (oder ein VARIANT, das ein OBJECT enthält) zurückzugeben.

• Verwenden Sie subField(<Index>), um ein Column-Objekt für ein Element in einem ARRAY (oder ein VARIANT, das ein ARRAY enthält) zurückzugeben.

Im folgenden Code wird beispielsweise das Feld dealership in Objekten der Spalte src der Beispieldaten ausgewählt:

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

Der Code gibt Folgendes aus:

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

Sie können auch Methodenaufrufe verketten, um einen Pfad zu einem bestimmten Feld oder Element zu durchlaufen.

Im folgende Code wird zum Beispiel das Feld name im Objekt salesperson ausgewählt:

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

Der Code gibt Folgendes aus:

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

In einem weiteren Codebeispiel wird das erste Element des Feldes vehicle ausgewählt, das eine Reihe von Fahrzeugen enthält. Im Beispiel wird auch das Feld price aus dem ersten Element ausgewählt.

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();

Der Code gibt Folgendes aus:

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

Als Alternative zur Methode apply können Sie die Funktionen get, get_ignore_case oder get_path verwenden, wenn der Feldname oder die Elemente im Pfad unregelmäßig sind und die Verwendung der Column.subField-Methoden erschweren.

Die folgenden Codezeilen geben zum Beispiel beide den Wert eines bestimmten Feldes in einem Objekt aus:

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

In ähnlicher Weise geben die folgenden Codezeilen beide den Wert eines Feldes für einen bestimmten Pfad in einem Objekt aus:

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();

Explizites Umwandeln von Werten semistrukturierter Daten¶

Standardmäßig werden die Werte von Feldern und Elementen als Zeichenfolgenliterale (einschließlich der doppelten Anführungszeichen) zurückgegeben, wie in den obigen Beispielen gezeigt.

Um unerwartete Ergebnisse zu vermeiden, rufen Sie die Methode cast auf, um den Wert in einen ganz bestimmten Typ umzuwandeln. Mit dem folgenden Code werden zum Beispiel die Werte ohne und mit Umwandlung ausgegeben:

// 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();

Der Code gibt Folgendes aus:

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

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

Vereinfachen eines Arrays von Objekten zu Zeilen¶

Wenn Sie semistrukturierte Daten für ein DataFrame vereinfachen müssen (z. B. eine Zeile für jedes Objekt in einem Array erstellen), rufen Sie die Methode flatten auf. Diese Methode entspricht der SQL-Funktion FLATTEN. Wenn Sie einen Pfad zu einem Objekt oder Array übergeben, gibt die Methode ein DataFrame zurück, das eine Zeile für jedes Feld oder Element im Objekt oder Array enthält.

In den Beispieldaten ist beispielsweise src:customer ein Array von Objekten, die Informationen über einen Kunden enthalten. Jedes Objekt enthält ein Feld name und ein Feld address.

Wenn Sie diesen Pfad an die Funktion flatten übergeben:

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