org.apache.crunch.kafka.offset.hdfs

Class HDFSOffsetWriter

java.lang.Object

org.apache.crunch.kafka.offset.AbstractOffsetWriter

org.apache.crunch.kafka.offset.hdfs.HDFSOffsetWriter

All Implemented Interfaces:

Closeable, AutoCloseable, OffsetWriter

public class HDFSOffsetWriter
extends AbstractOffsetWriter

Offset writer implementation that stores the offsets in HDFS.

Field Summary

Fields

Modifier and Type	Field and Description
static String	FILE_FORMAT_EXTENSION File extension for storing the offsets.
static org.joda.time.format.DateTimeFormatter	FILE_FORMATTER Formatter to use when creating the file names in a URI compliant format.
static String	PERSIST_TIME_FORMAT Custom formatter for translating the times into valid file names.

Constructor Summary

Constructors

Constructor and Description
HDFSOffsetWriter(org.apache.hadoop.conf.Configuration config, org.apache.hadoop.fs.Path baseStoragePath) Creates a writer instance for interacting with the storage specified by the config and with the base storage path of baseStoragePath.

Method Summary

Modifier and Type	Method and Description
void	close()
static long	fileNameToPersistenceTime(String fileName) Converts a fileName into the time the offsets were persisted.
static org.apache.hadoop.fs.Path	getPersistedTimeStoragePath(org.apache.hadoop.fs.Path baseStoragePath, long persistedTime) Creates a Path for storing the offsets for a specified persistedTime.
static String	persistenceTimeToFileName(long persistedTime) Converts a persistedTime into a file name for persisting the offsets.
void	write(long asOfTime, Map<org.apache.kafka.common.TopicPartition,Long> offsets) Persists the offsets to a configured location with metadata of asOfTime indicating the time in milliseconds when the offsets were meaningful.

Methods inherited from class org.apache.crunch.kafka.offset.AbstractOffsetWriter

write

Methods inherited from class java.lang.Object

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

Field Detail

PERSIST_TIME_FORMAT

public static final String PERSIST_TIME_FORMAT

Custom formatter for translating the times into valid file names.

See Also:

Constant Field Values

FILE_FORMATTER

public static final org.joda.time.format.DateTimeFormatter FILE_FORMATTER

Formatter to use when creating the file names in a URI compliant format.

FILE_FORMAT_EXTENSION

public static final String FILE_FORMAT_EXTENSION

File extension for storing the offsets.

See Also:

Constant Field Values

Constructor Detail

HDFSOffsetWriter

public HDFSOffsetWriter(org.apache.hadoop.conf.Configuration config,
                        org.apache.hadoop.fs.Path baseStoragePath)

Creates a writer instance for interacting with the storage specified by the config and with the base storage path of baseStoragePath.

Parameters:

config - the config for interacting with the underlying data store.

baseStoragePath - the base storage path for offset information.

Throws:

IllegalArgumentException - if either argument is null.

Method Detail

write

public void write(long asOfTime,
                  Map<org.apache.kafka.common.TopicPartition,Long> offsets)
           throws IOException

Description copied from interface: OffsetWriter

Persists the offsets to a configured location with metadata of asOfTime indicating the time in milliseconds when the offsets were meaningful.

Parameters:

asOfTime - the metadata describing when the offsets are accurate as of a time given in milliseconds since epoch.

offsets - the offsets to persist

Throws:

IOException - if there is an error persisting the offsets.

close

public void close()
           throws IOException

Throws:

IOException

getPersistedTimeStoragePath

public static org.apache.hadoop.fs.Path getPersistedTimeStoragePath(org.apache.hadoop.fs.Path baseStoragePath,
                                                                    long persistedTime)

Creates a Path for storing the offsets for a specified persistedTime.

Parameters:

baseStoragePath - The base path the offsets will be stored at.

persistedTime - the time of the data being persisted.

Returns:

The path to where the offset information should be stored.

Throws:

IllegalArgumentException - if the baseStoragePath is null.

fileNameToPersistenceTime

public static long fileNameToPersistenceTime(String fileName)

Converts a fileName into the time the offsets were persisted.

Parameters:

fileName - the file name to parse.

Returns:

the time in milliseconds since epoch that the offsets were stored.

Throws:

IllegalArgumentException - if the fileName is not of the correct format or is null or empty.

persistenceTimeToFileName

public static String persistenceTimeToFileName(long persistedTime)

Converts a persistedTime into a file name for persisting the offsets.

Parameters:

persistedTime - the persisted time to use to generate the file name.

Returns:

the file name to use when persisting the data.