Comment écrire le contenu d'un Scala de flux dans un fichier?

J'ai un Scala flux d'octets que je voudrais écrire dans un fichier. Le flux a trop de données à la mémoire tampon de l'ensemble de la mémoire.

Comme une première tentative, j'ai créé un InputStream similaire à ceci:

class MyInputStream(data: Stream[Byte]) extends InputStream {
  private val iterator = data.iterator
  override def read(): Int = if (iterator.hasNext) iterator.next else -1
}

Puis-je utiliser Apache Commons pour écrire le fichier:

val source = new MyInputStream(dataStream)
val target = new FileOutputStream(file)
try {
  IOUtils.copy(source, target)
} finally {
  target.close
}

Cela fonctionne, mais je ne suis pas trop satisfait de la performance. Je suppose que l'appel MyInputStream.read pour chaque octet introduit beaucoup de frais généraux. Est-il un meilleur moyen?

OriginalL'auteur mrog | 2015-04-30

  1. 5

    Vous pouvez (ou non!) se tromper que le côté lecture est la source de vos problèmes de performance. Il pourrait être le fait que vous êtes à l'aide d'un tampon FileOutputStream(...), en forçant un système distinct d'appel pour chaque octet écrit.

    Voici mon point de vue, quick 'n simple:

    def writeBytes( data : Stream[Byte], file : File ) = {
  val target = new BufferedOutputStream( new FileOutputStream(file) );
  try data.foreach( target.write(_) ) finally target.close;
}
    Bonne prise! J'ai défini une 100 MO test de flux: (1 to 100 * 1024 * 1024).toStream.map(_.toByte). Mise en mémoire tampon du flux de sortie amélioré le temps d'exécution moyen de 16,5 secondes à 14,7 secondes. Il n'est pas moi la performance, je veux, mais ça aide un peu. J'imagine qu'il serait d'aider plus si j'avais un disque dur mécanique au lieu d'un SSD.
    Je vais accepter ce que le vainqueur parce qu'il a produit les meilleurs résultats lors de mes tests. Il a été seulement légèrement mieux que la plupart des autres techniques, mais la différence était assez clair. J'ai appris que Stream[Byte] est vraiment un moyen inefficace pour gérer les données. Le ruisseau lui-même est le goulot d'étranglement majeur. Stream[String] ou Stream[Array[Byte]] sont de bien meilleurs choix pour déplacer rapidement les données.

    OriginalL'auteur Steve Waldman

  2. 5

    Je recommanderais le java.nio.file paquet. Avec Files.write vous pouvez écrire Arrays de Bytes à un Path construit à partir d'un nom de fichier.

    C'est à vous de la façon de fournir les Bytes. Vous pouvez tourner le Stream dans un Array avec .toArray ou vous pouvez take octets désactivé le service d'un (ou plusieurs) à la fois et de les transformer en tableaux.

    Ici est un simple bloc de code montrant la .toArray méthode.

    import java.nio.file.{Files, Paths}

val filename: String = "output.bin"
val bytes: Stream[Byte] = ...
Files.write(Paths.get(filename), bytes.toArray)
    Je pense que le problème @mrog est face à ce .toArray forces que le flux soit matérialisée dans la mémoire et il est trop grand pour cela.
    J'ai essayé le NIO approche, de manière récursive le fractionnement du flux pour obtenir les octets. Il a fallu 16 secondes pour écrire 100 MO, ce qui est presque identique dans les performances de l'original MyInputStream approche. La source que j'utilise est (1 to 100 * 1024 * 1024).toStream.map(_.toByte), qui devrait être beaucoup rapide. Par comparaison, en copiant les mêmes données à partir d'un fichier à l'autre à l'aide de Files.copy(Paths.get("/tmp/input"), Paths.get("/tmp/output"), StandardCopyOption.REPLACE_EXISTING) prend 0.19 secondes. Je me sens comme je suis absent dehors sur deux ordres de grandeur des performances.
    J'ai fait un test de performance sur MyInputStream, juste à parcourir les données, mais ne pas écrire n'importe où. 100 MO a pris 0.002 secondes. Ainsi, le test de flux que j'utilise n'est pas le goulot d'étranglement. Et à l'aide de IOUtils.copier pour copier les mêmes données d'un fichier à un autre prend 0,12 secondes (même plus rapide que d'utiliser NIO), de sorte que le système de fichiers n'est pas le goulot d'étranglement.
    J'ai raté son coup quand j'ai testé le flux de la performance. Itération sur les flux de matières (sans faire n'importe quoi avec les données) s'13.8 secondes, pas 0.002 secondes. Ainsi, le flux est le goulot d'étranglement.

    OriginalL'auteur Aphex

  3. 2

    Vous devez mettre en œuvre l'essentiel de lire remplacer dans votre InputStream mise en œuvre:

    override def read(b: Array[Byte], off: Int, len: Int)

    IOUtils.copy utilise que la signature de lire/écrire dans des blocs de 4 ko.

    J'ai essayé et n'ai pas remarqué de différence dans la performance. Peut-être que c'est parce que j'étais toujours en s'appuyant sur les flux de itérateur pour lire un octet à la fois. Pouvez-vous suggérer une bonne mise en œuvre de read(b: Array[Byte], off: Int, len: Int) qui ne repose pas sur un flux d'itérateur?

    OriginalL'auteur Arne Claassen

  4. 2

    Étant donné que StreamIterator de la lecture d'un octet à la fois peut être le goulot d'étranglement, j'ai conçu une méthode pour écrire un flux vers unOutputStream qui ne repose pas sur lui et il est à priori plus efficace:

    object StreamCopier {
  def copy(data: Stream[Byte], output: OutputStream) = {
    def write(d: Stream[Byte]): Unit = if (d.nonEmpty) {
      val (head, tail) = d.splitAt(4 * 1024)
      val bytes = head.toArray
      output.write(bytes, 0, bytes.length)
      write(tail)
    }
    write(data)
  }
}

    EDIT: correction d'un bug en remplaçant data avec d l'intérieur de la queue-récursive write fonction.

    Cette approche utilise une approche récursive via splitAt pour diviser le flux dans la première ~4K et le reste, écrire que la tête à la OutputStream et de manière récursive sur la queue de la rivière, jusqu'à ce que splitAt retourne un vide stream.

    Puisque vous avez des tests de performances, à la place, je vais le laisser à vous de juger si cela s'avère être plus efficace.

    J'aime cette approche, car il est assez simple. J'ai eu de refactoriser un peu parce que la méthode extérieure était de garder une référence à la rivière, tandis que la méthode intérieure couru, donc de consommer beaucoup de mémoire. objet StreamCopier2 { @tailrec def copie(données: Flux[Byte], sortie: OutputStream): Unit = { if (data.non vide) { val (tête, queue) = données.splitAt(4 * 1024) val octets = la tête.toArray de sortie.écrire(octets, 0, octets.la longueur) copie(queue, sortie) } } }
    Curieusement, la nouvelle version est exactement la même quantité de temps pour traiter 100 MO, mon premier MyInputStream approche. La différence majeure est que MyInputStream avantages légèrement de l'ajout d'un BufferedOutputStream, mais StreamCopier2 s'exécute dans le même laps de temps si il y a un BufferedOutputStream ou pas. Cela rend MyInputStream avec un BufferedOutputStream le gagnant par une petite marge.
    Désolé à ce sujet.. Que la fonction a été complètement brisée. Comme vous l'avez souligné data ne devrait pas avoir été utilisé à l'intérieur de write et puis je n'ai même jamais de coups de pied au large de la queue de la récursivité. Mon fix est un peu différent de votre suggestion, afin de voir si cela fait une différence pour vous.
    La version éditée fonctionne, mais il est lent pour les grands ruisseaux. Je crois data reste dans la portée jusqu'à write est fini, provoquant la JVM pour les déchets se rassemblent fréquemment. 1 MO stream prend 0,4 secondes, à 10 MB stream prend 5.1 secondes, ce qui est un peu plus augmenter que je ne l'aurais espérer. J'ai essayé de 100 MO stream, mais j'ai abandonné après de nombreuses minutes d'attente pour elle à la fin. Sans imbriquée méthodes, c'est beaucoup plus rapide.
    J'ai redémarré mon IDE et essaya de nouveau. 100 MB stream a pris 110 secondes, ce qui est près de 7 fois plus longtemps que le temps qu'il faut sans imbriquée méthodes.

    OriginalL'auteur Arne Claassen