org.apache.spark.storage

Class MemoryStore

Object

org.apache.spark.storage.BlockStore

org.apache.spark.storage.MemoryStore

All Implemented Interfaces:

Logging

public class MemoryStore
extends BlockStore

Stores blocks in memory, either as Arrays of deserialized Java objects or as serialized ByteBuffers.

Constructor Summary

Constructors

Constructor and Description
MemoryStore(BlockManager blockManager, long maxMemory)

Method Summary

Methods

Modifier and Type	Method and Description
void	clear()
boolean	contains(BlockId blockId)
long	currentUnrollMemory() Return the amount of memory currently occupied for unrolling blocks across all threads.
long	currentUnrollMemoryForThisThread() Return the amount of memory currently occupied for unrolling blocks by this thread.
long	freeMemory() Free memory not occupied by existing blocks.
scala.Option<java.nio.ByteBuffer>	getBytes(BlockId blockId)
long	getSize(BlockId blockId) Return the size of a block in bytes.
scala.Option<scala.collection.Iterator<Object>>	getValues(BlockId blockId)
void	logMemoryUsage() Log information about current memory usage.
void	logUnrollFailureMessage(BlockId blockId, long finalVectorSize) Log a warning for failing to unroll a block.
int	numThreadsUnrolling() Return the number of threads currently unrolling blocks.
PutResult	putArray(BlockId blockId, Object[] values, StorageLevel level, boolean returnValues)
PutResult	putBytes(BlockId blockId, java.nio.ByteBuffer _bytes, StorageLevel level)
PutResult	putIterator(BlockId blockId, scala.collection.Iterator<Object> values, StorageLevel level, boolean returnValues) Put in a block and, possibly, also return its content as either bytes or another Iterator.
PutResult	putIterator(BlockId blockId, scala.collection.Iterator<Object> values, StorageLevel level, boolean returnValues, boolean allowPersistToDisk) Attempt to put the given block in memory store.
void	releaseUnrollMemoryForThisThread(long memory) Release memory used by this thread for unrolling blocks.
boolean	remove(BlockId blockId) Remove a block, if it exists.
boolean	reserveUnrollMemoryForThisThread(long memory) Reserve additional memory for unrolling blocks used by this thread.
scala.util.Either<Object[],scala.collection.Iterator<Object>>	unrollSafely(BlockId blockId, scala.collection.Iterator<Object> values, scala.collection.mutable.ArrayBuffer<scala.Tuple2<BlockId,BlockStatus>> droppedBlocks) Unroll the given block in memory safely.

Methods inherited from class org.apache.spark.storage.BlockStore

blockManager

Methods inherited from class Object

equals, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Methods inherited from interface org.apache.spark.Logging

initializeIfNecessary, initializeLogging, isTraceEnabled, log_, log, logDebug, logDebug, logError, logError, logInfo, logInfo, logName, logTrace, logTrace, logWarning, logWarning

Constructor Detail

MemoryStore

public MemoryStore(BlockManager blockManager,
           long maxMemory)

Method Detail

freeMemory

public long freeMemory()

Free memory not occupied by existing blocks. Note that this does not include unroll memory.

getSize

public long getSize(BlockId blockId)

Description copied from class: BlockStore

Return the size of a block in bytes.

Specified by:

getSize in class BlockStore

putBytes

public PutResult putBytes(BlockId blockId,
                 java.nio.ByteBuffer _bytes,
                 StorageLevel level)

Specified by:

putBytes in class BlockStore

putArray

public PutResult putArray(BlockId blockId,
                 Object[] values,
                 StorageLevel level,
                 boolean returnValues)

Specified by:

putArray in class BlockStore

putIterator

public PutResult putIterator(BlockId blockId,
                    scala.collection.Iterator<Object> values,
                    StorageLevel level,
                    boolean returnValues)

Description copied from class: BlockStore

Put in a block and, possibly, also return its content as either bytes or another Iterator. This is used to efficiently write the values to multiple locations (e.g. for replication).

Specified by:

putIterator in class BlockStore

Returns:

a PutResult that contains the size of the data, as well as the values put if returnValues is true (if not, the result's data field can be null)

putIterator

public PutResult putIterator(BlockId blockId,
                    scala.collection.Iterator<Object> values,
                    StorageLevel level,
                    boolean returnValues,
                    boolean allowPersistToDisk)

Attempt to put the given block in memory store.

There may not be enough space to fully unroll the iterator in memory, in which case we optionally drop the values to disk if (1) the block's storage level specifies useDisk, and (2) allowPersistToDisk is true.

One scenario in which allowPersistToDisk is false is when the BlockManager reads a block back from disk and attempts to cache it in memory. In this case, we should not persist the block back on disk again, as it is already in disk store.

getBytes

public scala.Option<java.nio.ByteBuffer> getBytes(BlockId blockId)

Specified by:

getBytes in class BlockStore

getValues

public scala.Option<scala.collection.Iterator<Object>> getValues(BlockId blockId)

Specified by:

getValues in class BlockStore

remove

public boolean remove(BlockId blockId)

Description copied from class: BlockStore

Remove a block, if it exists.

Specified by:

remove in class BlockStore

Parameters:

blockId - the block to remove.

Returns:

True if the block was found and removed, False otherwise.

clear

public void clear()

Overrides:

clear in class BlockStore

unrollSafely

public scala.util.Either<Object[],scala.collection.Iterator<Object>> unrollSafely(BlockId blockId,
                                                                         scala.collection.Iterator<Object> values,
                                                                         scala.collection.mutable.ArrayBuffer<scala.Tuple2<BlockId,BlockStatus>> droppedBlocks)

Unroll the given block in memory safely.

The safety of this operation refers to avoiding potential OOM exceptions caused by unrolling the entirety of the block in memory at once. This is achieved by periodically checking whether the memory restrictions for unrolling blocks are still satisfied, stopping immediately if not. This check is a safeguard against the scenario in which there is not enough free memory to accommodate the entirety of a single block.

This method returns either an array with the contents of the entire block or an iterator containing the values of the block (if the array would have exceeded available memory).

contains

public boolean contains(BlockId blockId)

Specified by:

contains in class BlockStore

reserveUnrollMemoryForThisThread

public boolean reserveUnrollMemoryForThisThread(long memory)

Reserve additional memory for unrolling blocks used by this thread. Return whether the request is granted.

releaseUnrollMemoryForThisThread

public void releaseUnrollMemoryForThisThread(long memory)

Release memory used by this thread for unrolling blocks. If the amount is not specified, remove the current thread's allocation altogether.

currentUnrollMemory

public long currentUnrollMemory()

Return the amount of memory currently occupied for unrolling blocks across all threads.

currentUnrollMemoryForThisThread

public long currentUnrollMemoryForThisThread()

Return the amount of memory currently occupied for unrolling blocks by this thread.

numThreadsUnrolling

public int numThreadsUnrolling()

Return the number of threads currently unrolling blocks.

logMemoryUsage

public void logMemoryUsage()

Log information about current memory usage.

logUnrollFailureMessage

public void logUnrollFailureMessage(BlockId blockId,
                           long finalVectorSize)

Log a warning for failing to unroll a block.

Parameters:

blockId - ID of the block we are trying to unroll.

finalVectorSize - Final size of the vector before unrolling failed.