Megjegyzés
Az oldalhoz való hozzáféréshez engedély szükséges. Megpróbálhat bejelentkezni vagy módosítani a címtárat.
Az oldalhoz való hozzáféréshez engedély szükséges. Megpróbálhatja módosítani a címtárat.
A streamelt DataFrame külső tárolórendszerekből (például fájlrendszerekből és kulcs-érték tárolókból) való betöltéséhez használt felület. Ezt a hozzáférést használhatja spark.readStream .
Szemantika
# Access through SparkSession
spark.readStream
Metódusok
| Módszer | Leírás |
|---|---|
format(source) |
Megadja a bemeneti adatforrás formátumát. |
schema(schema) |
A streamelt DataFrame sémáját adja meg. |
option(key, value) |
Hozzáad egy bemeneti beállítást az alapul szolgáló adatforráshoz. |
options(**options) |
Több bemeneti lehetőséget is hozzáad a mögöttes adatforráshoz. |
load(path) |
Betölti a streamelési DataFrame-et a megadott elérési útról, és visszaadja. |
json(path) |
Betölt egy JSON-fájlstreamet, és egy DataFrame-et ad vissza. |
orc(path) |
Betölt egy ORC-fájlstreamet, és egy DataFrame-et ad vissza. |
parquet(path) |
Betölt egy Parquet-fájlstreamet, és egy DataFrame-et ad vissza. |
text(path) |
Betölt egy szöveges fájlstreamet, és egy DataFrame-et ad vissza. |
csv(path) |
Betölt egy CSV-fájlstreamet, és egy DataFrame-et ad vissza. |
xml(path) |
Betölt egy XML-fájlstreamet, és egy DataFrame-et ad vissza. |
table(tableName) |
Betölt egy streamelési Delta-táblát, és egy DataFrame-et ad vissza. |
name(source_name) |
Nevet rendel a streamforráshoz az ellenőrzőpont-fejlesztéshez. |
changes(tableName) |
A megadott tábla sorszintű módosításait (Adatrögzítés módosítása) adja vissza streamelési adatkeretként. |
Examples
spark.readStream
# <...streaming.readwriter.DataStreamReader object ...>
Töltsön be egy sebességfolyamot, alkalmazzon átalakítást, írjon a konzolra, és 3 másodperc után állítsa le.
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()