|
|
@ -5,7 +5,6 @@ import org.apache.hadoop.fs.*; |
|
|
|
import org.apache.hadoop.fs.permission.AclEntry; |
|
|
|
import org.apache.hadoop.fs.permission.AclStatus; |
|
|
|
import org.apache.hadoop.fs.permission.FsPermission; |
|
|
|
import org.apache.hadoop.security.AccessControlException; |
|
|
|
import org.apache.hadoop.security.UserGroupInformation; |
|
|
|
import org.apache.hadoop.util.Progressable; |
|
|
|
import org.slf4j.Logger; |
|
|
@ -14,20 +13,19 @@ import seaweedfs.client.FilerProto; |
|
|
|
|
|
|
|
import java.io.FileNotFoundException; |
|
|
|
import java.io.IOException; |
|
|
|
import java.io.InputStream; |
|
|
|
import java.io.OutputStream; |
|
|
|
import java.net.URI; |
|
|
|
import java.util.EnumSet; |
|
|
|
import java.util.List; |
|
|
|
import java.util.Map; |
|
|
|
|
|
|
|
import static org.apache.hadoop.fs.CommonConfigurationKeysPublic.IO_FILE_BUFFER_SIZE_KEY; |
|
|
|
|
|
|
|
public class SeaweedFileSystem extends FileSystem { |
|
|
|
|
|
|
|
public static final int FS_SEAWEED_DEFAULT_PORT = 8888; |
|
|
|
public static final String FS_SEAWEED_FILER_HOST = "fs.seaweed.filer.host"; |
|
|
|
public static final String FS_SEAWEED_FILER_PORT = "fs.seaweed.filer.port"; |
|
|
|
public static final int FS_SEAWEED_DEFAULT_PORT = 8888; |
|
|
|
public static final String FS_SEAWEED_BUFFER_SIZE = "fs.seaweed.buffer.size"; |
|
|
|
public static final int FS_SEAWEED_DEFAULT_BUFFER_SIZE = 4 * 1024 * 1024; |
|
|
|
|
|
|
|
private static final Logger LOG = LoggerFactory.getLogger(SeaweedFileSystem.class); |
|
|
|
|
|
|
@ -75,8 +73,9 @@ public class SeaweedFileSystem extends FileSystem { |
|
|
|
path = qualify(path); |
|
|
|
|
|
|
|
try { |
|
|
|
FSInputStream inputStream = seaweedFileSystemStore.openFileForRead(path, statistics, bufferSize); |
|
|
|
return new FSDataInputStream(new BufferedFSInputStream(inputStream, 16 * 1024 * 1024)); |
|
|
|
int seaweedBufferSize = this.getConf().getInt(FS_SEAWEED_BUFFER_SIZE, FS_SEAWEED_DEFAULT_BUFFER_SIZE); |
|
|
|
FSInputStream inputStream = seaweedFileSystemStore.openFileForRead(path, statistics, seaweedBufferSize); |
|
|
|
return new FSDataInputStream(new BufferedFSInputStream(inputStream, 4 * seaweedBufferSize)); |
|
|
|
} catch (Exception ex) { |
|
|
|
LOG.warn("open path: {} bufferSize:{}", path, bufferSize, ex); |
|
|
|
return null; |
|
|
@ -93,7 +92,8 @@ public class SeaweedFileSystem extends FileSystem { |
|
|
|
|
|
|
|
try { |
|
|
|
String replicaPlacement = String.format("%03d", replication - 1); |
|
|
|
OutputStream outputStream = seaweedFileSystemStore.createFile(path, overwrite, permission, bufferSize, replicaPlacement); |
|
|
|
int seaweedBufferSize = this.getConf().getInt(FS_SEAWEED_BUFFER_SIZE, FS_SEAWEED_DEFAULT_BUFFER_SIZE); |
|
|
|
OutputStream outputStream = seaweedFileSystemStore.createFile(path, overwrite, permission, seaweedBufferSize, replicaPlacement); |
|
|
|
return new FSDataOutputStream(outputStream, statistics); |
|
|
|
} catch (Exception ex) { |
|
|
|
LOG.warn("create path: {} bufferSize:{} blockSize:{}", path, bufferSize, blockSize, ex); |
|
|
@ -103,8 +103,9 @@ public class SeaweedFileSystem extends FileSystem { |
|
|
|
|
|
|
|
/** |
|
|
|
* {@inheritDoc} |
|
|
|
* |
|
|
|
* @throws FileNotFoundException if the parent directory is not present -or |
|
|
|
* is not a directory. |
|
|
|
* is not a directory. |
|
|
|
*/ |
|
|
|
@Override |
|
|
|
public FSDataOutputStream createNonRecursive(Path path, |
|
|
@ -121,9 +122,10 @@ public class SeaweedFileSystem extends FileSystem { |
|
|
|
throw new FileAlreadyExistsException("Not a directory: " + parent); |
|
|
|
} |
|
|
|
} |
|
|
|
int seaweedBufferSize = this.getConf().getInt(FS_SEAWEED_BUFFER_SIZE, FS_SEAWEED_DEFAULT_BUFFER_SIZE); |
|
|
|
return create(path, permission, |
|
|
|
flags.contains(CreateFlag.OVERWRITE), bufferSize, |
|
|
|
replication, blockSize, progress); |
|
|
|
replication, seaweedBufferSize, progress); |
|
|
|
} |
|
|
|
|
|
|
|
@Override |
|
|
@ -133,7 +135,8 @@ public class SeaweedFileSystem extends FileSystem { |
|
|
|
|
|
|
|
path = qualify(path); |
|
|
|
try { |
|
|
|
OutputStream outputStream = seaweedFileSystemStore.createFile(path, false, null, bufferSize, ""); |
|
|
|
int seaweedBufferSize = this.getConf().getInt(FS_SEAWEED_BUFFER_SIZE, FS_SEAWEED_DEFAULT_BUFFER_SIZE); |
|
|
|
OutputStream outputStream = seaweedFileSystemStore.createFile(path, false, null, seaweedBufferSize, ""); |
|
|
|
return new FSDataOutputStream(outputStream, statistics); |
|
|
|
} catch (Exception ex) { |
|
|
|
LOG.warn("append path: {} bufferSize:{}", path, bufferSize, ex); |
|
|
@ -338,9 +341,7 @@ public class SeaweedFileSystem extends FileSystem { |
|
|
|
|
|
|
|
@Override |
|
|
|
public void createSymlink(final Path target, final Path link, |
|
|
|
final boolean createParent) throws AccessControlException, |
|
|
|
FileAlreadyExistsException, FileNotFoundException, |
|
|
|
ParentNotDirectoryException, UnsupportedFileSystemException, |
|
|
|
final boolean createParent) throws |
|
|
|
IOException { |
|
|
|
// Supporting filesystems should override this method |
|
|
|
throw new UnsupportedOperationException( |
|
|
|