co.cask.cdap.data.stream

Class StreamInputFormat<K,V>

java.lang.Object

org.apache.hadoop.mapreduce.InputFormat<K,V>

co.cask.cdap.data.stream.StreamInputFormat<K,V>

Type Parameters:

K - Key type of input

V - Value type of input

public class StreamInputFormat<K,V>
extends InputFormat<K,V>

Stream mapreduce input format. Stream data files are organized by partition directories with bucket files inside.

Each file has path pattern

     [streamName]/[partitionName]/[bucketName].[dat|idx]
 OR
     [streamName]/[generation]/[partitionName]/[bucketName].[dat|idx]
 

Where .dat is the event data file, .idx is the accompany index file.

The generation is an integer, representing the stream generation of the data under it. When a stream is truncated, the generation increment by one. The generation 0 is a special case that there is no generation directory.

The partitionName is formatted as

   [partitionStartTime].[duration]
 

with both partitionStartTime and duration in seconds.

The bucketName is formatted as

   [prefix].[seqNo]
 

The seqNo is a strictly increasing integer for the same prefix starting with 0.

Constructor Summary

Constructors

Constructor and Description
StreamInputFormat()

Method Summary

Methods

Modifier and Type	Method and Description
RecordReader<K,V>	createRecordReader(InputSplit split, TaskAttemptContext context)
protected co.cask.cdap.api.stream.StreamEventDecoder<K,V>	createStreamEventDecoder(Configuration conf)
protected long	getCurrentTime()
static Class<? extends co.cask.cdap.api.stream.StreamEventDecoder>	getDecoderClass(Configuration conf) Returns the StreamEventDecoder class as specified in the job configuration.
List<InputSplit>	getSplits(JobContext context)
static void	inferDecoderClass(Configuration conf, Type vClass) Tries to set the DECODER_TYPE depending upon the supplied value class
static void	setBodyFormatSpecification(Configuration conf, co.cask.cdap.api.data.format.FormatSpecification formatSpecification) Set the format specification for reading the body of stream events.
static void	setDecoderClassName(Configuration conf, String decoderClassName) Sets the class name for the StreamEventDecoder.
static void	setMaxSplitSize(Configuration conf, long maxSplits) Sets the maximum split size.
static void	setMinSplitSize(Configuration conf, long minSplits) Sets the minimum split size.
static void	setStreamPath(Configuration conf, URI path) Sets the base path to stream files.
static void	setTimeRange(Configuration conf, long startTime, long endTime) Sets the time range for the stream events.
static void	setTTL(Configuration conf, long ttl) Sets the TTL for the stream events.

Methods inherited from class java.lang.Object

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

Constructor Detail

StreamInputFormat

public StreamInputFormat()

Method Detail

setTTL

public static void setTTL(Configuration conf,
          long ttl)

Sets the TTL for the stream events.

Parameters:

conf - The configuration to modify

ttl - TTL of the stream in milliseconds.

setTimeRange

public static void setTimeRange(Configuration conf,
                long startTime,
                long endTime)

Sets the time range for the stream events.

Parameters:

conf - The configuration to modify

startTime - Timestamp in milliseconds of the event start time (inclusive).

endTime - Timestamp in milliseconds of the event end time (exclusive).

setStreamPath

public static void setStreamPath(Configuration conf,
                 URI path)

Sets the base path to stream files.

Parameters:

conf - The conf to modify.

path - The file path to stream base directory.

setMaxSplitSize

public static void setMaxSplitSize(Configuration conf,
                   long maxSplits)

Sets the maximum split size.

Parameters:

conf - The conf to modify.

maxSplits - Maximum split size in bytes.

setMinSplitSize

public static void setMinSplitSize(Configuration conf,
                   long minSplits)

Sets the minimum split size.

Parameters:

conf - The conf to modify.

minSplits - Minimum split size in bytes.

setDecoderClassName

public static void setDecoderClassName(Configuration conf,
                       String decoderClassName)

Sets the class name for the StreamEventDecoder.

Parameters:

conf - The conf to modify.

decoderClassName - Class name of the decoder class

getDecoderClass

public static Class<? extends co.cask.cdap.api.stream.StreamEventDecoder> getDecoderClass(Configuration conf)

Returns the StreamEventDecoder class as specified in the job configuration.

Parameters:

conf - The job configuration

Returns:

The StreamEventDecoder class or null if it is not set.

setBodyFormatSpecification

public static void setBodyFormatSpecification(Configuration conf,
                              co.cask.cdap.api.data.format.FormatSpecification formatSpecification)

Set the format specification for reading the body of stream events. Will also set the decoder class appropriately.

Parameters:

conf - The job configuration.

formatSpecification - Format specification for reading the body of stream events.

inferDecoderClass

public static void inferDecoderClass(Configuration conf,
                     Type vClass)

Tries to set the DECODER_TYPE depending upon the supplied value class

Parameters:

conf - the conf to modify

vClass - the value class Type

getSplits

public List<InputSplit> getSplits(JobContext context)
                           throws IOException,
                                  InterruptedException

Specified by:

getSplits in class InputFormat<K,V>

Throws:

IOException

InterruptedException

createRecordReader

public RecordReader<K,V> createRecordReader(InputSplit split,
                                   TaskAttemptContext context)
                                     throws IOException,
                                            InterruptedException

Specified by:

createRecordReader in class InputFormat<K,V>

Throws:

IOException

InterruptedException

getCurrentTime

protected long getCurrentTime()

createStreamEventDecoder

protected co.cask.cdap.api.stream.StreamEventDecoder<K,V> createStreamEventDecoder(Configuration conf)