Lettura di file molto grandi (~ 1 TB) in blocchi sequenziali

Question

Ho bisogno di leggere un file di grandi dimensioni in Scala ed elaborarlo in blocchi di k bit (k potrebbe essere 65536 in genere). Come un semplice esempio (ma non quello che voglio):

i blocchi di file sono (f1, f2, ... fk).

voglio calcolare SHA256(f1)+SHA256(f2)+...+ SHA256(fk)

Tale calcolo può essere effettuata gradualmente con unica costante stoccaggio e il blocco corrente senza dover altri blocchi.

Qual è il modo migliore per leggere il file? (forse qualcosa che utilizza le continuazioni?)

MODIFICA: il tipo di domanda collegata risolve il problema ma non sempre, poiché il file che sto guardando contiene dati binari.

Answer 1

Ecco un approccio che utilizza Akka Streams. Questo utilizza la memoria costante e può elaborare i blocchi di file man mano che vengono letti.

Vedere "File di streaming I/O" nella parte inferiore di questa pagina per ulteriori informazioni. http://doc.akka.io/docs/akka-stream-and-http-experimental/1.0-RC3/scala/stream-io.html

Inizia con un semplice file build.sbt:

scalaVersion := "2.11.6" 

libraryDependencies ++= Seq(
     "com.typesafe.akka" %% "akka-stream-experimental" % "1.0-RC3" 
) 

le parti interessanti sono il Source, Flow e Sink. Il Source è un SynchronousFileSource che legge in un file di grandi dimensioni con una dimensione del blocco di 65536. Un valore ByteString della dimensione del blocco viene emesso dallo Source e consumato da uno Flow che calcola un hash SHA256 per ogni blocco. Infine, lo Sink consuma l'output dallo Flow e stampa gli array di byte. Dovrai convertirli e sommarli usando un fold per ottenere una somma totale.

import akka.stream.io._ 
import java.io.File 
import scala.concurrent.Future 
import akka.stream.scaladsl._ 
import akka.actor.ActorSystem 
import akka.stream.ActorFlowMaterializer 
import java.security.MessageDigest 

object LargeFile extends App{ 
    implicit val system = ActorSystem("Sys") 
    import system.dispatcher 
    implicit val materializer = ActorFlowMaterializer() 

    val file = new File("<path to large file>") 

    val fileSource = SynchronousFileSource(file, 65536) 

    val shaFlow = fileSource.map(chunk => sha256(chunk.toString)) 

    shaFlow.to(Sink.foreach(println(_))).run//TODO - Convert the byte[] and sum them using fold 

    def sha256(s: String) = { 
    val messageDigest = MessageDigest.getInstance("SHA-256") 
    messageDigest.digest(s.getBytes("UTF-8")) 
    } 
} 

BYTE ARRAYS!

> run 
[info] Running LargeFile 
[B@3d0587a6 
[B@360cc296 
[B@7fbb2192 
... 

Answer 2

Creazione del digest utilizzando il flusso continuo, che credo produce un iteratore

import java.File 
import java.FileInputStream 
import java.security.MessageDigest 

val file = new File("test.in") 
val is = new FileInputStream(file) 

val md = MessageDigest.getInstance("SHA-256") 

val bytes = Array.fill[Byte](65536)(0) 

Stream 
    .continually((is.read(bytes),bytes)) 
    .takeWhile(_._1 != -1) 
    .foreach{ x => md.update(x._2,0,x._1) } 

println(md.digest()) 
// prinln(md.digest().map("%02X" format _).mkString) // if you want hex string