Comment enregistrer l'agrégation en flux continu dans le mode de sortie Complet vers le parquet?

Question

J'ai appliqué l'agrégation sur la base de données en streaming en utilisant le mode complet. Pour enregistrer les données en local, j'ai implémenté foreach sink. Je suis capable de sauvegarder des données sous forme de texte. Mais j'ai besoin de le sauvegarder en format parquet.

val writerForText = new ForeachWriter[Row] { 
    var fileWriter: FileWriter = _ 

    override def process(value: Row): Unit = { 
     fileWriter.append(value.toSeq.mkString(",")) 
    } 

    override def close(errorOrNull: Throwable): Unit = { 
     fileWriter.close() 
    } 

    override def open(partitionId: Long, version: Long): Boolean = { 
     FileUtils.forceMkdir(new File(s"src/test/resources/${partitionId}")) 
     fileWriter = new FileWriter(new File(s"src/test/resources/${partitionId}/temp")) 
     true 

    } 
    } 

val columnName = "col1" 
frame.select(count(columnName),count(columnName),min(columnName),mean(columnName),max(columnName),first(columnName), last(columnName), sum(columnName)) 
       .writeStream.outputMode(OutputMode.Complete()).foreach(writerForText).start() 

Comment puis-je y parvenir? Merci d'avance!

Answer 1

Pour enregistrer les données en local, j'ai implémenté foreach sink. Je suis capable de sauvegarder des données sous forme de texte. Mais j'ai besoin de le sauvegarder en format parquet.

Le format par défaut lors de l'enregistrement d'un dataset streaming est ... parquet. Cela dit, vous n'avez pas besoin d'utiliser un évier foreach assez avancé, mais simplement parquet.

La requête pourrait être comme suit:

scala> :type in 
org.apache.spark.sql.DataFrame 

scala> in.isStreaming 
res0: Boolean = true 

in.writeStream. 
    option("checkpointLocation", "/tmp/checkpoint-so"). 
    start("/tmp/parquets")