gobblin.util

Class HadoopUtils

java.lang.Object

gobblin.util.HadoopUtils

public class HadoopUtils
extends Object

A utility class for working with Hadoop.

Field Summary

Fields

Modifier and Type	Field and Description
static Collection<String>	FS_SCHEMES_NON_ATOMIC A Collection of all known FileSystem schemes that do not support atomic renames or copies.
static String	HDFS_ILLEGAL_TOKEN_REGEX
static String	MAX_FILESYSTEM_QPS

Constructor Summary

Constructors

Constructor and Description
HadoopUtils()

Method Summary

Modifier and Type	Method and Description
static void	addGobblinSite() Add "gobblin-site.xml" as a Configuration resource.
static void	copyFile(FileSystem srcFs, Path src, FileSystem dstFs, Path dst, boolean overwrite, Configuration conf) Copy a file from a srcFs FileSystem to a dstFs FileSystem.
static void	copyFile(FileSystem srcFs, Path src, FileSystem dstFs, Path dst, Path tmp, boolean overwriteDst, Configuration conf) Copies a src Path from a srcFs FileSystem to a dst Path on a dstFs FileSystem.
static void	copyPath(FileSystem srcFs, Path src, FileSystem dstFs, Path dst, boolean overwrite, Configuration conf) Copies data from a src Path to a dst Path.
static void	copyPath(FileSystem srcFs, Path src, FileSystem dstFs, Path dst, Configuration conf) Copies data from a src Path to a dst Path.
static void	deleteIfExists(FileSystem fs, Path path, boolean recursive) A wrapper around FileSystem.delete(Path, boolean) that only deletes a given Path if it is present on the given FileSystem.
static void	deletePath(FileSystem fs, Path f, boolean recursive) A wrapper around FileSystem.delete(Path, boolean) which throws IOException if the given Path exists, and FileSystem.delete(Path, boolean) returns False.
static void	deletePathAndEmptyAncestors(FileSystem fs, Path f, boolean recursive)
static Writable	deserializeFromString(Class<? extends Writable> writableClass, String serializedWritableStr) Deserialize a Writable object from a string.
static Writable	deserializeFromString(Class<? extends Writable> writableClass, String serializedWritableStr, Configuration configuration) Deserialize a Writable object from a string.
static FsPermission	deserializeFsPermission(State props, String propName, FsPermission defaultPermission) Get FsPermission from a State object.
static FsPermission	deserializeWriterDirPermissions(State state, int numBranches, int branchId) Deserializes a FsPermissions object that should be used when a DataWriter is creating directories.
static FsPermission	deserializeWriterFilePermissions(State state, int numBranches, int branchId) Deserializes a FsPermissions object that should be used when a DataWriter is writing a file.
static Configuration	getConfFromProperties(Properties properties)
static Configuration	getConfFromState(State state)
static Configuration	getConfFromState(State state, Optional<String> encryptedPath) Provides Hadoop configuration given state.
static FileSystem	getOptionallyThrottledFileSystem(FileSystem fs, int qpsLimit) Get a throttled FileSystem that limits the number of queries per second to a FileSystem.
static FileSystem	getOptionallyThrottledFileSystem(FileSystem fs, State state) Calls getOptionallyThrottledFileSystem(FileSystem, int) parsing the qps from the input State at key MAX_FILESYSTEM_QPS.
static State	getStateFromConf(Configuration conf)
static boolean	hasContent(FileSystem fs, Path path)
static List<FileStatus>	listStatusRecursive(FileSystem fileSystem, Path path) Deprecated. Use FileListUtils.listFilesRecursively(FileSystem, Path).
static void	movePath(FileSystem srcFs, Path src, FileSystem dstFs, Path dst, boolean overwrite, Configuration conf) Moves a src Path from a srcFs FileSystem to a dst Path on a dstFs FileSystem.
static void	movePath(FileSystem srcFs, Path src, FileSystem dstFs, Path dst, Configuration conf) Moves a src Path from a srcFs FileSystem to a dst Path on a dstFs FileSystem.
static Configuration	newConfiguration()
static void	renamePath(FileSystem fs, Path oldName, Path newName) A wrapper around FileSystem.rename(Path, Path) which throws IOException if FileSystem.rename(Path, Path) returns False.
static void	renamePath(FileSystem fs, Path oldName, Path newName, boolean overwrite) A wrapper around FileSystem.rename(Path, Path) which throws IOException if FileSystem.rename(Path, Path) returns False.
static boolean	renamePathHandleLocalFSRace(FileSystem fs, Path src, Path dst) Renames a src Path on fs FileSystem to a dst Path.
static void	renameRecursively(FileSystem fileSystem, Path from, Path to) This method is an additive implementation of the FileSystem.rename(Path, Path) method.
static boolean	safeRenameIfNotExists(FileSystem fs, Path from, Path to) Renames from to to if to doesn't exist in a thread-safe way.
static void	safeRenameRecursively(FileSystem fileSystem, Path from, Path to) A thread safe variation of renamePath(FileSystem, Path, Path) which can be used in multi-threaded/multi-mapper environment.
static Path	sanitizePath(Path path, String substitute) Remove illegal HDFS path characters from the given path.
static String	sanitizePath(String path, String substitute) Remove illegal HDFS path characters from the given path.
static String	serializeToString(Writable writable) Serialize a Writable object into a string.
static void	serializeWriterDirPermissions(State state, int numBranches, int branchId, FsPermission fsPermissions) Given a FsPermission objects, set a key, value pair in the given State for the writer to use when creating files.
static void	serializeWriterFilePermissions(State state, int numBranches, int branchId, FsPermission fsPermissions) Given a FsPermission objects, set a key, value pair in the given State for the writer to use when creating files.
static void	setGroup(FileSystem fs, Path path, String group) Set the group associated with a given path.
static void	setPermissions(Path location, Optional<String> owner, Optional<String> group, FileSystem fs, FsPermission permission) Try to set owner and permissions for the path.
static void	setWriterDirOctalPermissions(State state, int numBranches, int branchId, String octalPermissions) Given a String in octal notation, set a key, value pair in the given State for the writer to use when creating directories.
static void	setWriterFileOctalPermissions(State state, int numBranches, int branchId, String octalPermissions) Given a String in octal notation, set a key, value pair in the given State for the writer to use when creating files.
static String	toUriPath(Path path) Get the path as a string without schema or authority.
static boolean	unsafeRenameIfNotExists(FileSystem fs, Path from, Path to) Renames from to to if to doesn't exist in a non-thread-safe way.

Methods inherited from class java.lang.Object

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

Field Detail

HDFS_ILLEGAL_TOKEN_REGEX

public static final String HDFS_ILLEGAL_TOKEN_REGEX

See Also:

Constant Field Values

FS_SCHEMES_NON_ATOMIC

public static final Collection<String> FS_SCHEMES_NON_ATOMIC

A Collection of all known FileSystem schemes that do not support atomic renames or copies.

The following important properties are useful to remember when writing code that is compatible with S3:

• Renames are not atomic, and require copying the entire source file to the destination file
• Writes to S3 using FileSystem.create(Path) will first go to the local filesystem, when the stream is closed the local file will be uploaded to S3

MAX_FILESYSTEM_QPS

public static final String MAX_FILESYSTEM_QPS

See Also:

Constant Field Values

Constructor Detail

HadoopUtils

public HadoopUtils()

Method Detail

newConfiguration

public static Configuration newConfiguration()

listStatusRecursive

@Deprecated
public static List<FileStatus> listStatusRecursive(FileSystem fileSystem,
                                                               Path path)
                                                        throws IOException

Deprecated. Use FileListUtils.listFilesRecursively(FileSystem, Path).

Throws:

IOException

toUriPath

public static String toUriPath(Path path)

Get the path as a string without schema or authority. E.g. Converts sftp://user/data/file.txt to /user/data/file.txt

deletePath

public static void deletePath(FileSystem fs,
                              Path f,
                              boolean recursive)
                       throws IOException

A wrapper around FileSystem.delete(Path, boolean) which throws IOException if the given Path exists, and FileSystem.delete(Path, boolean) returns False.

Throws:

IOException

deleteIfExists

public static void deleteIfExists(FileSystem fs,
                                  Path path,
                                  boolean recursive)
                           throws IOException

A wrapper around FileSystem.delete(Path, boolean) that only deletes a given Path if it is present on the given FileSystem.

Throws:

IOException

deletePathAndEmptyAncestors

public static void deletePathAndEmptyAncestors(FileSystem fs,
                                               Path f,
                                               boolean recursive)
                                        throws IOException

Throws:

IOException

renamePathHandleLocalFSRace

public static boolean renamePathHandleLocalFSRace(FileSystem fs,
                                                  Path src,
                                                  Path dst)
                                           throws IOException

Renames a src Path on fs FileSystem to a dst Path. If fs is a LocalFileSystem and src is a directory then File.renameTo(java.io.File) is called directly to avoid a directory rename race condition where RawLocalFileSystem.rename(org.apache.hadoop.fs.Path, org.apache.hadoop.fs.Path) copies the conflicting src directory into dst resulting in an extra nested level, such as /root/a/b/c/e/e where e is repeated.

Parameters:

fs - the FileSystem where the src Path exists

src - the source Path which will be renamed

dst - the Path to rename to

Returns:

true if rename succeeded, false if rename failed.

Throws:

IOException - if rename failed for reasons other than target exists.

renamePath

public static void renamePath(FileSystem fs,
                              Path oldName,
                              Path newName)
                       throws IOException

A wrapper around FileSystem.rename(Path, Path) which throws IOException if FileSystem.rename(Path, Path) returns False.

Throws:

IOException

renamePath

public static void renamePath(FileSystem fs,
                              Path oldName,
                              Path newName,
                              boolean overwrite)
                       throws IOException

A wrapper around FileSystem.rename(Path, Path) which throws IOException if FileSystem.rename(Path, Path) returns False.

Throws:

IOException

movePath

public static void movePath(FileSystem srcFs,
                            Path src,
                            FileSystem dstFs,
                            Path dst,
                            Configuration conf)
                     throws IOException

Moves a src Path from a srcFs FileSystem to a dst Path on a dstFs FileSystem. If the srcFs and the dstFs have the same scheme, and neither of them or S3 schemes, then the Path is simply renamed. Otherwise, the data is from the src Path to the dst Path. So this method can handle copying data between different FileSystem implementations.

Parameters:

srcFs - the source FileSystem where the src Path exists

src - the source Path which will me moved

dstFs - the destination FileSystem where the dst Path should be created

dst - the Path to move data to

Throws:

IOException

movePath

public static void movePath(FileSystem srcFs,
                            Path src,
                            FileSystem dstFs,
                            Path dst,
                            boolean overwrite,
                            Configuration conf)
                     throws IOException

Moves a src Path from a srcFs FileSystem to a dst Path on a dstFs FileSystem. If the srcFs and the dstFs have the same scheme, and neither of them or S3 schemes, then the Path is simply renamed. Otherwise, the data is from the src Path to the dst Path. So this method can handle copying data between different FileSystem implementations.

Parameters:

srcFs - the source FileSystem where the src Path exists

src - the source Path which will me moved

dstFs - the destination FileSystem where the dst Path should be created

dst - the Path to move data to

overwrite - true if the destination should be overwritten; otherwise, false

Throws:

IOException

copyPath

public static void copyPath(FileSystem srcFs,
                            Path src,
                            FileSystem dstFs,
                            Path dst,
                            Configuration conf)
                     throws IOException

Copies data from a src Path to a dst Path.

This method should be used in preference to FileUtil.copy(FileSystem, Path, FileSystem, Path, boolean, boolean, Configuration), which does not handle clean up of incomplete files if there is an error while copying data.

TODO this method does not handle cleaning up any local files leftover by writing to S3.

Parameters:

srcFs - the source FileSystem where the src Path exists

src - the Path to copy from the source FileSystem

dstFs - the destination FileSystem where the dst Path should be created

dst - the Path to copy data to

Throws:

IOException

copyPath

public static void copyPath(FileSystem srcFs,
                            Path src,
                            FileSystem dstFs,
                            Path dst,
                            boolean overwrite,
                            Configuration conf)
                     throws IOException

Copies data from a src Path to a dst Path.

This method should be used in preference to FileUtil.copy(FileSystem, Path, FileSystem, Path, boolean, boolean, Configuration), which does not handle clean up of incomplete files if there is an error while copying data.

TODO this method does not handle cleaning up any local files leftover by writing to S3.

Parameters:

srcFs - the source FileSystem where the src Path exists

src - the Path to copy from the source FileSystem

dstFs - the destination FileSystem where the dst Path should be created

dst - the Path to copy data to

overwrite - true if the destination should be overwritten; otherwise, false

Throws:

IOException

copyFile

public static void copyFile(FileSystem srcFs,
                            Path src,
                            FileSystem dstFs,
                            Path dst,
                            Path tmp,
                            boolean overwriteDst,
                            Configuration conf)
                     throws IOException

Copies a src Path from a srcFs FileSystem to a dst Path on a dstFs FileSystem. If either the srcFs or dstFs are S3 FileSystems (as dictated by FS_SCHEMES_NON_ATOMIC) then data is directly copied from the src to the dst. Otherwise data is first copied to a tmp Path, which is then renamed to the dst.

Parameters:

srcFs - the source FileSystem where the src Path exists

src - the Path to copy from the source FileSystem

dstFs - the destination FileSystem where the dst Path should be created

dst - the Path to copy data to

tmp - the temporary Path to use when copying data

overwriteDst - true if the destination and tmp path should should be overwritten, false otherwise

Throws:

IOException

copyFile

public static void copyFile(FileSystem srcFs,
                            Path src,
                            FileSystem dstFs,
                            Path dst,
                            boolean overwrite,
                            Configuration conf)
                     throws IOException

Copy a file from a srcFs FileSystem to a dstFs FileSystem. The src Path must be a file, that is FileSystem.isFile(Path) must return true for src.

If overwrite is specified to true, this method may delete the dst directory even if the copy from src to dst fails.

Parameters:

srcFs - the src FileSystem to copy the file from

src - the src Path to copy

dstFs - the destination FileSystem to write to

dst - the destination Path to write to

overwrite - true if the dst Path should be overwritten, false otherwise

Throws:

IOException

renameRecursively

public static void renameRecursively(FileSystem fileSystem,
                                     Path from,
                                     Path to)
                              throws IOException

This method is an additive implementation of the FileSystem.rename(Path, Path) method. It moves all the files/directories under 'from' path to the 'to' path without overwriting existing directories in the 'to' path.

The rename operation happens at the first non-existent sub-directory. If a directory at destination path already exists, it recursively tries to move sub-directories. If all the sub-directories also exist at the destination, a file level move is done

Parameters:

fileSystem - on which the data needs to be moved

from - path of the data to be moved

to - path of the data to be moved

Throws:

IOException

getOptionallyThrottledFileSystem

public static FileSystem getOptionallyThrottledFileSystem(FileSystem fs,
                                                          State state)
                                                   throws IOException

Calls getOptionallyThrottledFileSystem(FileSystem, int) parsing the qps from the input State at key MAX_FILESYSTEM_QPS.

Throws:

IOException

getOptionallyThrottledFileSystem

public static FileSystem getOptionallyThrottledFileSystem(FileSystem fs,
                                                          int qpsLimit)
                                                   throws IOException

Get a throttled FileSystem that limits the number of queries per second to a FileSystem. If the input qps is <= 0, no such throttling will be performed.

Throws:

IOException

safeRenameIfNotExists

public static boolean safeRenameIfNotExists(FileSystem fs,
                                            Path from,
                                            Path to)
                                     throws IOException

Renames from to to if to doesn't exist in a thread-safe way. This method is necessary because FileSystem.rename(org.apache.hadoop.fs.Path, org.apache.hadoop.fs.Path) is inconsistent across file system implementations, e.g. in some of them rename(foo, bar) will create bar/foo if bar already existed, but it will only create bar if it didn't.

The thread-safety is only guaranteed among calls to this method. An external modification to the relevant target directory could still cause unexpected results in the renaming.

Parameters:

fs - filesystem where rename will be executed.

from - origin Path.

to - target Path.

Returns:

true if rename succeeded, false if the target already exists.

Throws:

IOException - if rename failed for reasons other than target exists.

unsafeRenameIfNotExists

public static boolean unsafeRenameIfNotExists(FileSystem fs,
                                              Path from,
                                              Path to)
                                       throws IOException

Renames from to to if to doesn't exist in a non-thread-safe way.

Parameters:

fs - filesystem where rename will be executed.

from - origin Path.

to - target Path.

Returns:

true if rename succeeded, false if the target already exists.

Throws:

IOException - if rename failed for reasons other than target exists.

safeRenameRecursively

public static void safeRenameRecursively(FileSystem fileSystem,
                                         Path from,
                                         Path to)
                                  throws IOException

A thread safe variation of renamePath(FileSystem, Path, Path) which can be used in multi-threaded/multi-mapper environment. The rename operation always happens at file level hence directories are not overwritten under the 'to' path.

If the contents of destination 'to' path is not expected to be modified concurrently, use renamePath(FileSystem, Path, Path) which is faster and more optimized

NOTE: This does not seem to be working for all FileSystem implementations. Use renameRecursively(FileSystem, Path, Path)

Parameters:

fileSystem - on which the data needs to be moved

from - path of the data to be moved

to - path of the data to be moved

Throws:

IOException

getConfFromState

public static Configuration getConfFromState(State state)

getConfFromState

public static Configuration getConfFromState(State state,
                                             Optional<String> encryptedPath)

Provides Hadoop configuration given state. It also supports decrypting values on "encryptedPath". Note that this encryptedPath path will be removed from full path of each config key and leaving only child path on the key(s). If there's same config path as child path, the one stripped will have higher priority. e.g: - encryptedPath: writer.fs.encrypted before: writer.fs.encrypted.secret after: secret Common use case for these encryptedPath: When there's have encrypted credential in job property but you'd like Filesystem to get decrypted value.

Parameters:

srcConfig - source config.

encryptedPath - Optional. If provided, config that is on this path will be decrypted. @see ConfigUtils.resolveEncrypted Note that config on encryptedPath will be included in the end result even it's not part of includeOnlyPath

Returns:

Hadoop Configuration.

getConfFromProperties

public static Configuration getConfFromProperties(Properties properties)

getStateFromConf

public static State getStateFromConf(Configuration conf)

setGroup

public static void setGroup(FileSystem fs,
                            Path path,
                            String group)
                     throws IOException

Set the group associated with a given path.

Parameters:

fs - the FileSystem instance used to perform the file operation

path - the given path

group - the group associated with the path

Throws:

IOException

serializeToString

public static String serializeToString(Writable writable)
                                throws IOException

Serialize a Writable object into a string.

Parameters:

writable - the Writable object to be serialized

Returns:

a string serialized from the Writable object

Throws:

IOException - if there's something wrong with the serialization

deserializeFromString

public static Writable deserializeFromString(Class<? extends Writable> writableClass,
                                             String serializedWritableStr)
                                      throws IOException

Deserialize a Writable object from a string.

Parameters:

writableClass - the Writable implementation class

serializedWritableStr - the string containing a serialized Writable object

Returns:

a Writable deserialized from the string

Throws:

IOException - if there's something wrong with the deserialization

deserializeFromString

public static Writable deserializeFromString(Class<? extends Writable> writableClass,
                                             String serializedWritableStr,
                                             Configuration configuration)
                                      throws IOException

Deserialize a Writable object from a string.

Parameters:

writableClass - the Writable implementation class

serializedWritableStr - the string containing a serialized Writable object

configuration - a Configuration object containing Hadoop configuration properties

Returns:

a Writable deserialized from the string

Throws:

IOException - if there's something wrong with the deserialization

serializeWriterFilePermissions

public static void serializeWriterFilePermissions(State state,
                                                  int numBranches,
                                                  int branchId,
                                                  FsPermission fsPermissions)

Given a FsPermission objects, set a key, value pair in the given State for the writer to use when creating files. This method should be used in conjunction with deserializeWriterFilePermissions(State, int, int).

serializeWriterDirPermissions

public static void serializeWriterDirPermissions(State state,
                                                 int numBranches,
                                                 int branchId,
                                                 FsPermission fsPermissions)

Given a FsPermission objects, set a key, value pair in the given State for the writer to use when creating files. This method should be used in conjunction with deserializeWriterDirPermissions(State, int, int).

setWriterFileOctalPermissions

public static void setWriterFileOctalPermissions(State state,
                                                 int numBranches,
                                                 int branchId,
                                                 String octalPermissions)

Given a String in octal notation, set a key, value pair in the given State for the writer to use when creating files. This method should be used in conjunction with deserializeWriterFilePermissions(State, int, int).

setWriterDirOctalPermissions

public static void setWriterDirOctalPermissions(State state,
                                                int numBranches,
                                                int branchId,
                                                String octalPermissions)

Given a String in octal notation, set a key, value pair in the given State for the writer to use when creating directories. This method should be used in conjunction with deserializeWriterDirPermissions(State, int, int).

deserializeWriterFilePermissions

public static FsPermission deserializeWriterFilePermissions(State state,
                                                            int numBranches,
                                                            int branchId)

Deserializes a FsPermissions object that should be used when a DataWriter is writing a file.

deserializeWriterDirPermissions

public static FsPermission deserializeWriterDirPermissions(State state,
                                                           int numBranches,
                                                           int branchId)

Deserializes a FsPermissions object that should be used when a DataWriter is creating directories.

deserializeFsPermission

public static FsPermission deserializeFsPermission(State props,
                                                   String propName,
                                                   FsPermission defaultPermission)

Get FsPermission from a State object.

Parameters:

props - A State containing properties.

propName - The property name for the permission. If not contained in the given state, defaultPermission will be used.

defaultPermission - default permission if propName is not contained in props.

Returns:

An FsPermission object.

sanitizePath

public static String sanitizePath(String path,
                                  String substitute)

Remove illegal HDFS path characters from the given path. Illegal characters will be replaced with the given substitute.

sanitizePath

public static Path sanitizePath(Path path,
                                String substitute)

Remove illegal HDFS path characters from the given path. Illegal characters will be replaced with the given substitute.

setPermissions

public static void setPermissions(Path location,
                                  Optional<String> owner,
                                  Optional<String> group,
                                  FileSystem fs,
                                  FsPermission permission)

Try to set owner and permissions for the path. Will not throw exception.

hasContent

public static boolean hasContent(FileSystem fs,
                                 Path path)
                          throws IOException

Throws:

IOException

addGobblinSite

public static void addGobblinSite()

Add "gobblin-site.xml" as a Configuration resource.