DataStreamWriter

Interface utilisée pour écrire un DataFrame de diffusion en continu vers des systèmes de stockage externes (par exemple, des systèmes de fichiers et des magasins clé-valeur). Permet df.writeStream d’y accéder.

Syntaxe

# Access through DataFrame
df.writeStream

Méthodes

Méthode Description
outputMode(outputMode) Spécifie la façon dont les données d’un DataFrame de streaming sont écrites dans le récepteur. Les options disponibles sont les suivantes : append, complete et update.
format(source) Spécifie le format de source de données de sortie.
option(key, value) Ajoute une option de sortie pour la source de données sous-jacente.
options(**options) Ajoute plusieurs options de sortie pour la source de données sous-jacente.
partitionBy(*cols) Partitionne la sortie par les colonnes données sur le système de fichiers.
clusterBy(*cols) Clusters la sortie par les colonnes données.
queryName(queryName) Spécifie le nom de la requête de diffusion en continu.
trigger(**kwargs) Définit le déclencheur pour l’exécution de la requête de diffusion en continu.
foreach(f) Définit la sortie de la requête de diffusion en continu à traiter par la fonction ou l’objet donné.
foreachBatch(func) Définit la sortie de chaque microbatch à traiter par la fonction donnée.
start(path) Démarre l’exécution de la requête de diffusion en continu et retourne un StreamingQuery objet.
table(tableName) Alias de toTable(). Écrit des données dans la table spécifiée et retourne un StreamingQuery objet.
toTable(tableName) Démarre l’exécution de la requête de diffusion en continu, en mettant continuellement les résultats dans la table donnée.

Exemples

Chargez un flux de débit, appliquez une transformation, écrivez dans la console et arrêtez après 3 secondes.

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()
  • Last updated on