epclust/R/de_serialize.R

   1 #' (De)Serialization of a [big]matrix or data stream
   2 #'
   3 #' \code{binarize()} serializes a matrix or CSV file with minimal overhead, into a
   4 #' binary file. \code{getDataInFile()} achieves the inverse task: she retrieves (ASCII)
   5 #' data rows from indices in the binary file. Finally, \code{binarizeTransform()}
   6 #' serialize transformations of all data chunks. To use it a data-retrieval function
   7 #' must be provided -- thus \code{binarize} will most likely be used first
   8 #' (and then a function defined to seek in generated binary file)
   9 #'
  10 #' @param data_ascii Either a matrix (by columns) or CSV file or connection (by rows)
  11 #' @param data_bin_file Name of binary file on output of (\code{binarize})
  12 #'   or input of (\code{getDataInFile})
  13 #' @param nb_per_chunk Number of lines to process in one batch (big.matrix or connection)
  14 #' @param getData Function to retrieve data chunks
  15 #' @param transform Transformation function to apply on data chunks
  16 #' @param indices Indices of the lines to retrieve
  17 #' @inheritParams claws
  18 #'
  19 #' @return For \code{getDataInFile()}, the matrix with rows corresponding to the
  20 #'   requested indices. \code{binarizeTransform} returns the number of processed lines.
  21 #'   \code{binarize} is designed to serialize in several calls, thus returns nothing.
  22 #'
  23 #' @name de_serialize
  24 #' @rdname de_serialize
  25 #' @aliases binarize binarizeTransform getDataInFile
  26 NULL
  27
  28 #' @rdname de_serialize
  29 #' @export
  30 binarize = function(data_ascii, data_bin_file, nb_per_chunk,
  31     sep=",", nbytes=4, endian=.Platform$endian)
  32 {
  33     # data_ascii can be of two types: [big.]matrix, or connection
  34     if (is.character(data_ascii))
  35         data_ascii = file(data_ascii, open="r")
  36     else if (methods::is(data_ascii,"connection") && !isOpen(data_ascii))
  37         open(data_ascii)
  38     is_matrix = !methods::is(data_ascii,"connection")
  39
  40     # At first call, the length of a stored row is written. So it's important to determine
  41     # if the serialization process already started.
  42     first_write = (!file.exists(data_bin_file) || file.info(data_bin_file)$size == 0)
  43
  44     # Open the binary file for writing (or 'append' if already exists)
  45     data_bin = file(data_bin_file, open=ifelse(first_write,"wb","ab"))
  46
  47     if (first_write)
  48     {
  49         # Write data length on first call: number of items always on 8 bytes
  50         writeBin(0L, data_bin, size=8, endian=endian)
  51         if (is_matrix)
  52             data_length = nrow(data_ascii)
  53         else #connection
  54         {
  55             # Read the first line to know data length, and write it then
  56             data_line = scan(data_ascii, double(), sep=sep, nlines=1, quiet=TRUE)
  57             writeBin(data_line, data_bin, size=nbytes, endian=endian)
  58             data_length = length(data_line)
  59         }
  60     }
  61
  62     if (is_matrix)
  63     {
  64         # Data is processed by chunks; although this may not be so useful for (normal) matrix
  65         # input, it could for a file-backed big.matrix. It's easier to follow a unified pattern.
  66         index = 1
  67     }
  68     repeat
  69     {
  70         if (is_matrix)
  71         {
  72             data_chunk =
  73                 if (index <= ncol(data_ascii))
  74                     as.double(data_ascii[,index:min(ncol(data_ascii),index+nb_per_chunk-1)])
  75                 else
  76                     double(0)
  77             index = index + nb_per_chunk
  78         }
  79         else #connection
  80             data_chunk = scan(data_ascii, double(), sep=sep, nlines=nb_per_chunk, quiet=TRUE)
  81
  82         # Data size is unknown in the case of a connection
  83         if (length(data_chunk)==0)
  84             break
  85
  86         # Write this chunk of data to the binary file
  87         writeBin(data_chunk, data_bin, size=nbytes, endian=endian)
  88     }
  89
  90     if (first_write)
  91     {
  92         # Write data_length, = (file_size-1) / (nbytes*nbWritten) at offset 0 in data_bin
  93         ignored = seek(data_bin, 0)
  94         writeBin(data_length, data_bin, size=8, endian=endian)
  95     }
  96     close(data_bin)
  97
  98     if ( ! is_matrix )
  99         close(data_ascii)
 100 }
 101
 102 #' @rdname de_serialize
 103 #' @export
 104 binarizeTransform = function(getData, transform, data_bin_file, nb_per_chunk,
 105     nbytes=4, endian=.Platform$endian)
 106 {
 107     nb_items = 0 #side-effect: store the number of transformed items
 108     index = 1
 109     repeat
 110     {
 111         # Retrieve a chunk of data in a binary file (generally obtained by binarize())
 112         data_chunk = getData((index-1)+seq_len(nb_per_chunk))
 113         if (is.null(data_chunk))
 114             break
 115
 116         # Apply transformation on the current chunk (by columns)
 117         transformed_chunk = transform(data_chunk)
 118
 119         # Save the result in binary format
 120         binarize(transformed_chunk, data_bin_file, nb_per_chunk, ",", nbytes, endian)
 121
 122         index = index + nb_per_chunk
 123         nb_items = nb_items + ncol(data_chunk)
 124     }
 125     nb_items #number of transformed items
 126 }
 127
 128 #' @rdname de_serialize
 129 #' @export
 130 getDataInFile = function(indices, data_bin_file, nbytes=4, endian=.Platform$endian)
 131 {
 132     data_bin = file(data_bin_file, "rb") #source binary file
 133
 134     data_size = file.info(data_bin_file)$size #number of bytes in the file
 135     # data_length: length of a vector in the binary file (first element, 8 bytes)
 136     data_length = readBin(data_bin, "integer", n=1, size=8, endian=endian)
 137
 138     # Seek all 'indices' columns in the binary file, using data_length and nbytes
 139     # to compute the offset ( index i at 8 + i*data_length*nbytes )
 140     data_ascii = do.call( cbind, lapply( indices, function(i) {
 141         offset = 8+(i-1)*data_length*nbytes
 142         if (offset >= data_size)
 143             return (NULL)
 144         ignored = seek(data_bin, offset) #position cursor at computed offset
 145         readBin(data_bin, "double", n=data_length, size=nbytes, endian=endian)
 146     } ) )
 147     close(data_bin)
 148
 149     data_ascii #retrieved data, in columns
 150 }