io.druid.data.input.impl.prefetch

Class PrefetchableTextFilesFirehoseFactory<T>

java.lang.Object

io.druid.data.input.impl.AbstractTextFilesFirehoseFactory<T>

io.druid.data.input.impl.prefetch.PrefetchableTextFilesFirehoseFactory<T>

All Implemented Interfaces:

FirehoseFactory<StringInputRowParser>

Direct Known Subclasses:

HttpFirehoseFactory, StaticAzureBlobStoreFirehoseFactory, StaticCloudFilesFirehoseFactory, StaticGoogleBlobStoreFirehoseFactory, StaticS3FirehoseFactory

public abstract class PrefetchableTextFilesFirehoseFactory<T>
extends AbstractTextFilesFirehoseFactory<T>

PrefetchableTextFilesFirehoseFactory is an abstract firehose factory for reading text files. The firehose returned by this class provides three key functionalities.

- Caching: for the first call of connect(StringInputRowParser, File), it caches objects in a local disk up to maxCacheCapacityBytes. These caches are NOT deleted until the process terminates, and thus can be used for future reads.
- Fetching: when it reads all cached data, it fetches remaining objects into a local disk and reads data from them. For the performance reason, prefetch technique is used, that is, when the size of remaining fetched data is smaller than prefetchTriggerBytes, a background prefetch thread automatically starts to fetch remaining objects.
- Retry: if an exception occurs while downloading an object, it retries again up to maxFetchRetry.

This implementation can be useful when the cost for reading input objects is large as reading from AWS S3 because batch tasks like IndexTask or HadoopIndexTask can read the whole data twice for determining partition specs and generating segments if the intervals of GranularitySpec is not specified.
Prefetching can be turned on/off by setting maxFetchCapacityBytes. Depending on prefetching is enabled or disabled, the behavior of the firehose is different like below.

1. If prefetch is enabled, this firehose can fetch input objects in background.
2. When next() is called, it first checks that there are already fetched files in local storage.
2.1 If exists, it simply chooses a fetched file and returns a LineIterator reading that file.
2.2 If there is no fetched files in local storage but some objects are still remained to be read, the firehose fetches one of input objects in background immediately. If an IOException occurs while downloading the object, it retries up to the maximum retry count. Finally, the firehose returns a LineIterator only when the download operation is successfully finished.
3. If prefetch is disabled, the firehose returns a LineIterator which directly reads the stream opened by AbstractTextFilesFirehoseFactory.openObjectStream(T). If there is an IOException, it will throw it and the read will fail.

Constructor Summary

Constructors

Constructor and Description
PrefetchableTextFilesFirehoseFactory(Long maxCacheCapacityBytes, Long maxFetchCapacityBytes, Long prefetchTriggerBytes, Long fetchTimeout, Integer maxFetchRetry)

Method Summary

Modifier and Type	Method and Description
Firehose	connect(StringInputRowParser firehoseParser, File temporaryDirectory) Initialization method that connects up the fire hose.
long	getFetchTimeout()
long	getMaxCacheCapacityBytes()
long	getMaxFetchCapacityBytes()
int	getMaxFetchRetry()
long	getPrefetchTriggerBytes()

Methods inherited from class io.druid.data.input.impl.AbstractTextFilesFirehoseFactory

initObjects, openObjectStream, wrapObjectStream

Methods inherited from class java.lang.Object

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

Methods inherited from interface io.druid.data.input.FirehoseFactory

connect

Constructor Detail

PrefetchableTextFilesFirehoseFactory

public PrefetchableTextFilesFirehoseFactory(Long maxCacheCapacityBytes,
                                            Long maxFetchCapacityBytes,
                                            Long prefetchTriggerBytes,
                                            Long fetchTimeout,
                                            Integer maxFetchRetry)

Method Detail

getMaxCacheCapacityBytes

public long getMaxCacheCapacityBytes()

getMaxFetchCapacityBytes

public long getMaxFetchCapacityBytes()

getPrefetchTriggerBytes

public long getPrefetchTriggerBytes()

getFetchTimeout

public long getFetchTimeout()

getMaxFetchRetry

public int getMaxFetchRetry()

connect

public Firehose connect(StringInputRowParser firehoseParser,
                        File temporaryDirectory)
                 throws IOException

Description copied from interface: FirehoseFactory

Initialization method that connects up the fire hose. If this method returns successfully it should be safe to call hasMore() on the returned Firehose (which might subsequently block).

If this method returns null, then any attempt to call hasMore(), nextRow(), commit() and close() on the return value will throw a surprising NPE. Throwing IOException on connection failure or runtime exception on invalid configuration is preferred over returning null.

Some fire hoses like PrefetchableTextFilesFirehoseFactory may use a temporary directory to cache data in it.

Specified by:

connect in interface FirehoseFactory<StringInputRowParser>

Overrides:

connect in class AbstractTextFilesFirehoseFactory<T>

Parameters:

firehoseParser - an input row parser

temporaryDirectory - a directory where temporary files are stored

Throws:

IOException