hdfs

README

hdfs-rest

封装 HDFS 客户端, 使用 HDFS Restful 接口操作 HDFS.

API Reference

API Reference : Directory Operations

• Mkdir: 创建目录
• MkdirAll: 创建目录

API Reference : File Operations

• Create: 按照默认参数创建文件
• CreateFile: 按照传入参数创建文件
• Append: 追加数据到指定文件
• Read: 读取整个文件
• ReadFile: 读取文件的某段数据

API Reference : Status Operations

• ListStatus: 当前路径下的所有文件(所有类型的文件)
• GetFileStatus: 获取指定文件状态
• Walk: 遍历指定路径下的所有文件(所有类型的文件)
• FileStatus.IsDir 判断是否为目录

Documentation

Index

• Constants
• Variables
• func GetErrFromBody(response *http.Response) error
• type Client
  • func New(opts ...Option) *Client
  • func (c *Client) Append(path string, bufferSize uint, data []byte) error
  • func (c *Client) Create(path string, data []byte) error
  • func (c *Client) CreateFile(path string, overwrite bool, blockSize uint64, perm os.FileMode, ...) error
  • func (c *Client) CreateFileWithGoodBlock(path string, data []byte) error
  • func (c *Client) GetContentSummary(dirname string) (*ContentSummary, error)
  • func (c *Client) GetFileChecksum(path string) (*FileChecksum, error)
  • func (c *Client) GetFileStatus(path string) (*FileStatus, error)
  • func (c *Client) ListStatus(path string) ([]FileStatus, error)
  • func (c *Client) Mkdir(dirname string, perm os.FileMode) error
  • func (c *Client) MkdirAll(dirname string, perm os.FileMode) error
  • func (c *Client) Read(path string) ([]byte, error)
  • func (c *Client) ReadFile(path string, offset, length uint64, bufferSize uint) ([]byte, error)
  • func (c *Client) RemoveAll(path string) error
  • func (c *Client) Walk(path string, callback func(prefixPath string, status FileStatus)) error
• type ContentSummary
• type FileChecksum
• type FileStatus
  • func (f FileStatus) IsDir() bool
• type Option
  • func HostOption(host string) Option
  • func MinBlockSizeOption(blockSize uint64) Option
  • func PortOption(port int) Option
  • func ReplicationOption(replication int) Option
  • func SSLOption(SSL bool) Option
  • func TimeoutOption(timeout time.Duration) Option
  • func TransportOption(transport *http.Transport) Option
  • func UserOption(user string) Option

Constants

const (
	DefaultUser        = "root" //默认用户
	DefaultReplication = 1      //默认复制数量
	DefaultProtocol    = "http"
	DefaultHost        = "127.0.0.1:14000"
	DefaultPort        = 14000
	DefaultBlockSize   = 1024 * 1024 * 128 //默认块大小
)

const (
	MkdirPathFormat         = "%s/webhdfs/v1%s?op=MKDIRS&permission=%s&user.name=%s"
	GetContentSummaryFormat = "%s/webhdfs/v1%s?op=GETCONTENTSUMMARY&user.name=%s"
)

const (
	ReadFileFormat        = "%s/webhdfs/v1%s?op=OPEN&length=%d&offset=%d&buffersize=%d&user.name=%s"
	DefaultReadFileFormat = "%s/webhdfs/v1%s?op=OPEN&user.name=%s"
	DefaultReadLength     = 4096
)

const (
	ListStatusFormat      = "%s/webhdfs/v1%s?op=LISTSTATUS&user.name=%s"
	GetFileStatusFormat   = "%s/webhdfs/v1%s?op=GETFILESTATUS&user.name=%s"
	GetFileChecksumFormat = "%s/webhdfs/v1%s?op=GETFILECHECKSUM&user.name=%s"
	TypeFile              = "FILE"
	TypeDir               = "DIRECTORY"
)

const (
	DefaultCreateFileFormat = "%s/webhdfs/v1%s?op=CREATE&overwrite=false&permission=644&buffersize=4096&replication=%d&user.name=%s"
	CreateFileFormat        = "%s/webhdfs/v1%s?op=CREATE&overwrite=%v&blocksize=%d&permission=%s&buffersize=%d&replication=%d&user.name=%s"
	AppendFileFormat        = "%s/webhdfs/v1%s?op=APPEND&buffersize=%d&user.name=%s"
	DefaultBufferSize       = 4096
	DefaultFilePerm         = 0644
	ContentTypeKey          = "Content-Type"
	ContentTypeValue        = "application/octet-stream"
)

const MB = 1024 * 1024

const (
	RemoveAllFormat = "%s/webhdfs/v1%s?op=DELETE&user.name=%s"
)

Variables

var (
	DefaultTransport = &http.Transport{
		Proxy: http.ProxyFromEnvironment,
		DialContext: (&net.Dialer{
			Timeout:   3 * time.Second,
			KeepAlive: 30 * time.Second,
		}).DialContext,
		ForceAttemptHTTP2:     true,
		MaxIdleConns:          100,
		IdleConnTimeout:       90 * time.Second,
		TLSHandshakeTimeout:   10 * time.Second,
		ExpectContinueTimeout: 1 * time.Second,
		TLSClientConfig: &tls.Config{
			InsecureSkipVerify: true,
		},
	}
)

var (
	ErrPathIsNotDirectory = fmt.Errorf("传入路径不是目录")
)

Functions

func GetErrFromBody

func GetErrFromBody(response *http.Response) error

Types

type Client

type Client struct {
	*http.Client
	// contains filtered or unexported fields
}

func New

func New(opts ...Option) *Client

func (*Client) Append

func (c *Client) Append(path string, bufferSize uint, data []byte) error

func (*Client) Create

func (c *Client) Create(path string, data []byte) error

Create 使用默认配置创建文件

func (*Client) CreateFile

func (c *Client) CreateFile(path string, overwrite bool, blockSize uint64, perm os.FileMode, bufferSize uint, replication uint16, data []byte) error

CreateFile 自定义配置创建文件

func (*Client) CreateFileWithGoodBlock

func (c *Client) CreateFileWithGoodBlock(path string, data []byte) error

CreateFileWithGoodBlock 使用合适的块大小存放数据

func (*Client) GetContentSummary

func (c *Client) GetContentSummary(dirname string) (*ContentSummary, error)

GetContentSummary 获取目录的内容摘要

func (*Client) GetFileChecksum

func (c *Client) GetFileChecksum(path string) (*FileChecksum, error)

GetFileChecksum 获取指定文件的md5值

func (*Client) GetFileStatus

func (c *Client) GetFileStatus(path string) (*FileStatus, error)

GetFileStatus 获取指定文件的信息

func (*Client) ListStatus

func (c *Client) ListStatus(path string) ([]FileStatus, error)

ListStatus 获取指定路径下列表

func (*Client) Mkdir

func (c *Client) Mkdir(dirname string, perm os.FileMode) error

Mkdir 创建目录并设置目录权限

func (*Client) MkdirAll

func (c *Client) MkdirAll(dirname string, perm os.FileMode) error

MkdirAll 创建目录并设置目录权限

func (*Client) Read

func (c *Client) Read(path string) ([]byte, error)

ReadFile 读取指定文件

func (*Client) ReadFile

func (c *Client) ReadFile(path string, offset, length uint64, bufferSize uint) ([]byte, error)

ReadFile 读取指定文件

func (*Client) RemoveAll

func (c *Client) RemoveAll(path string) error

func (*Client) Walk

func (c *Client) Walk(path string, callback func(prefixPath string, status FileStatus)) error

Walk 遍历某个路径下的目录

type ContentSummary

type ContentSummary struct {
	DirectoryCount int64
	FileCount      int64
	Length         int64
	Quota          int64
	SpaceConsumed  int64
	SpaceQuota     int64
}

type FileChecksum

type FileChecksum struct {
	Algorithm string
	Bytes     string
	Length    int64
}

type FileStatus

type FileStatus struct {
	AccesTime        int64
	BlockSize        int64
	Group            string
	Length           int64
	ModificationTime int64
	Owner            string
	PathSuffix       string
	Permission       string
	Replication      int64
	Type             string
}

func (FileStatus) IsDir

func (f FileStatus) IsDir() bool

type Option

type Option func(client *Client)

func HostOption

func HostOption(host string) Option

HostOption 主机

func MinBlockSizeOption

func MinBlockSizeOption(blockSize uint64) Option

MinBlockSizeOption 设置最小块大小 单位字节

func PortOption

func PortOption(port int) Option

PortOption 端口

func ReplicationOption

func ReplicationOption(replication int) Option

ReplicationOption 复制份数

func SSLOption

func SSLOption(SSL bool) Option

SSLOption use http or https. ture is https.

func TimeoutOption

func TimeoutOption(timeout time.Duration) Option

TimeoutOption 设置超时时间

func TransportOption

func TransportOption(transport *http.Transport) Option

TransportOption 设置传输

func UserOption

func UserOption(user string) Option

UserOption HDFS User