Hinweis
Für den Zugriff auf diese Seite ist eine Autorisierung erforderlich. Sie können versuchen, sich anzumelden oder das Verzeichnis zu wechseln.
Für den Zugriff auf diese Seite ist eine Autorisierung erforderlich. Sie können versuchen, das Verzeichnis zu wechseln.
Schnittstelle zum Laden eines Streaming-DataFrames aus externen Speichersystemen (z. B. Dateisysteme und Schlüsselwertspeicher). Hier können Sie spark.readStream darauf zugreifen.
Syntax
# Access through SparkSession
spark.readStream
Methodik
| Methode | Beschreibung |
|---|---|
format(source) |
Gibt das Format der Eingabedatenquelle an. |
schema(schema) |
Gibt das Schema des Streaming DataFrame an. |
option(key, value) |
Fügt eine Eingabeoption für die zugrunde liegende Datenquelle hinzu. |
options(**options) |
Fügt mehrere Eingabeoptionen für die zugrunde liegende Datenquelle hinzu. |
load(path) |
Lädt das Streaming-DataFrame aus dem angegebenen Pfad und gibt es zurück. |
json(path) |
Lädt einen JSON-Dateidatenstrom und gibt einen DataFrame zurück. |
orc(path) |
Lädt einen ORC-Dateidatenstrom und gibt einen DataFrame zurück. |
parquet(path) |
Lädt einen Geparkten Dateidatenstrom und gibt einen DataFrame zurück. |
text(path) |
Lädt einen Textdateidatenstrom und gibt einen DataFrame zurück. |
csv(path) |
Lädt einen CSV-Dateidatenstrom und gibt einen DataFrame zurück. |
xml(path) |
Lädt einen XML-Dateidatenstrom und gibt einen DataFrame zurück. |
table(tableName) |
Lädt eine Streaming-Delta-Tabelle und gibt einen DataFrame zurück. |
name(source_name) |
Weist der Streamingquelle einen Namen für die Prüfpunktentwicklung zu. |
changes(tableName) |
Gibt Änderungen auf Zeilenebene (Change Data Capture) aus der angegebenen Tabelle als Streaming DataFrame zurück. |
Beispiele
spark.readStream
# <...streaming.readwriter.DataStreamReader object ...>
Laden Sie einen Geschwindigkeitsstream, wenden Sie eine Transformation an, schreiben Sie die Konsole, und beenden Sie nach 3 Sekunden.
import time
df = spark.readStream.format("rate").load()
df = df.selectExpr("value % 3 as v")
q = df.writeStream.format("console").start()
time.sleep(3)
q.stop()