データストリームリーダー

外部ストレージ システム (ファイル システムやキー/値ストアなど) からストリーミングDataFrameロードするために使用されるインターフェイス。 これにアクセスするにはspark.readStreamを使用してください。

構文

Python

# Access through SparkSession
spark.readStream

方法

手法	説明
format(source)	入力データソースのフォーマットを指定します。
schema(schema)	ストリーミングDataFrameのスキーマを指定します。
option(key, value)	基となるデータソースへの入力オプションを追加します。
options(**options)	基となるデータソースに複数の入力オプションを追加します。
load(path)	指定されたパスからストリーミングDataFrameをロードして返します。
json(path)	JSONファイルストリームを読み込み、DataFrameを返します。
orc(path)	ORCファイルストリームを読み込み、DataFrameを返します。
parquet(path)	Parquetファイルストリームを読み込み、DataFrameを返します。
text(path)	テキストファイルストリームを読み込み、DataFrameを返します。
csv(path)	CSVファイルストリームを読み込み、DataFrameを返します。
xml(path)	XMLファイルストリームを読み込み、DataFrameを返します。
table(tableName)	ストリーミングDeltaテーブルをロードし、 DataFrameを返します。
name(source_name)	チェックポイント進化のためのストリーミングソースに名前を割り当てます。
changes(tableName)	指定されたテーブルから行レベルの変更 (変更データキャプチャ) をストリーミングDataFrameとして返します。

例

Python

spark.readStream
# <...streaming.readwriter.DataStreamReader object ...>

レートストリームを読み込み、変換を適用し、コンソールに書き込み、3秒後に停止します。

Python

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