Not
Bu sayfaya erişim yetkilendirme gerektiriyor. Oturum açmayı veya dizinleri değiştirmeyi deneyebilirsiniz.
Bu sayfaya erişim yetkilendirme gerektiriyor. Dizinleri değiştirmeyi deneyebilirsiniz.
Dış depolama sistemlerinden (örneğin, dosya sistemleri ve anahtar-değer depoları) akış DataFrame yüklemek için kullanılan arabirim. Buna erişmek için kullanın spark.readStream .
Sözdizimi
# Access through SparkSession
spark.readStream
Methods
| Yöntem | Açıklama |
|---|---|
format(source) |
Giriş veri kaynağı biçimini belirtir. |
schema(schema) |
Akış DataFrame şemasını belirtir. |
option(key, value) |
Temel alınan veri kaynağı için bir giriş seçeneği ekler. |
options(**options) |
Temel alınan veri kaynağı için birden çok giriş seçeneği ekler. |
load(path) |
Verilen yoldan akış DataFrame'i yükler ve döndürür. |
json(path) |
Bir JSON dosya akışını yükler ve bir DataFrame döndürür. |
orc(path) |
BIR ORC dosya akışını yükler ve bir DataFrame döndürür. |
parquet(path) |
Parquet dosya akışını yükler ve bir DataFrame döndürür. |
text(path) |
Bir metin dosyası akışı yükler ve bir DataFrame döndürür. |
csv(path) |
CSV dosya akışını yükler ve bir DataFrame döndürür. |
xml(path) |
XML dosya akışını yükler ve bir DataFrame döndürür. |
table(tableName) |
Akış Delta tablosunu yükler ve bir DataFrame döndürür. |
name(source_name) |
Denetim noktası evrimi için akış kaynağına bir ad atar. |
changes(tableName) |
Belirtilen tablodan satır düzeyi değişiklikleri (Veri Yakalamayı Değiştir) akış DataFrame olarak döndürür. |
Örnekler
spark.readStream
# <...streaming.readwriter.DataStreamReader object ...>
Hız akışı yükleyin, dönüştürme uygulayın, konsola yazın ve 3 saniye sonra durdurun.
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()