Observação
O acesso a essa página exige autorização. Você pode tentar entrar ou alterar diretórios.
O acesso a essa página exige autorização. Você pode tentar alterar os diretórios.
Interface usada para carregar um DataFrame de streaming de sistemas de armazenamento externos (por exemplo, sistemas de arquivos e repositórios de chave-valor). Use spark.readStream para acessar isso.
Sintaxe
# Access through SparkSession
spark.readStream
Methods
| Método | Descrição |
|---|---|
format(source) |
Especifica o formato de fonte de dados de entrada. |
schema(schema) |
Especifica o esquema do DataFrame de streaming. |
option(key, value) |
Adiciona uma opção de entrada para a fonte de dados subjacente. |
options(**options) |
Adiciona várias opções de entrada para a fonte de dados subjacente. |
load(path) |
Carrega o DataFrame de streaming do caminho fornecido e o retorna. |
json(path) |
Carrega um fluxo de arquivos JSON e retorna um DataFrame. |
orc(path) |
Carrega um fluxo de arquivos ORC e retorna um DataFrame. |
parquet(path) |
Carrega um fluxo de arquivos Parquet e retorna um DataFrame. |
text(path) |
Carrega um fluxo de arquivo de texto e retorna um DataFrame. |
csv(path) |
Carrega um fluxo de arquivos CSV e retorna um DataFrame. |
xml(path) |
Carrega um fluxo de arquivos XML e retorna um DataFrame. |
table(tableName) |
Carrega uma tabela Delta de streaming e retorna um DataFrame. |
name(source_name) |
Atribui um nome à fonte de streaming para a evolução do ponto de verificação. |
changes(tableName) |
Retorna alterações no nível da linha (Captura de Dados de Alteração) da tabela especificada como um DataFrame de streaming. |
Exemplos
spark.readStream
# <...streaming.readwriter.DataStreamReader object ...>
Carregue um fluxo de taxa, aplique uma transformação, escreva no console e pare após 3 segundos.
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()