当前位置：首页 > news >正文

node 与 webhdfs 交互

news 来源：原创 2024/9/20 8:55:34

注意 hadoop v3 之后，webhdfs 的端口为 9870 不是 50070

请求地址举例，例如创建一个文件：`host:9870/webhdfs/v1${hdfsFilePath}?op=CREATE&overwrite=true&user.name=username`

node 中使用。简单封装示例

import axios from 'axios'import HdfsConfig from './config.js';
import appConfig from '../config.js';const { hdfsAttachPath, hdfsPagePdfPath, localPagePdfPath, localAttachPath } = appConfig
const { host, port, path: hdfsPath, user: hdfsUser } = HdfsConfigimport { getCurrentDateFormatted } from '../utils.js';class WebHDFSClient {constructor() {this.host = host; // host e.g http://192.168.1.1this.port = port; // 9870this.user = hdfsUser;this.baseUrl = `http://${host}:${port}${hdfsPath}`; // http://192.168.1.1:9870/webhdfs/v1this.originPagePdfPath = '' // 当天的页面快照存储路径this.originAttachPath = '' // 当天的附件文件存储路径this.initDir()}// 列出目录async listDirectory (path) {const url = `${this.baseUrl}${path}?op=LISTSTATUS&user.name=${this.user}`;try {const response = await axios.get(url);return response.data.FileStatuses.FileStatus;} catch (error) {throw new Error(`Error listing directory: ${error.message}`);}}// 创建目录async createDirectory (path) {const url = `${this.baseUrl}${path}?op=MKDIRS&user.name=${this.user}`;try {const response = await axios.put(url);return response.data;} catch (error) {throw new Error(`Error creating directory: ${error.message}`);}}// 写入文件async writeFile (path, data) {const createUrl = `${this.baseUrl}${path}?op=CREATE&user.name=${this.user}&overwrite=true`;try {await axios({method: 'PUT',url: createUrl,headers: {'Content-Type': 'application/octet-stream', // 根据你的文件类型设置// 如果需要认证，添加相应的认证 headers，如 Authorization: Basic ...},data, // 创建文件流并作为请求体发送})console.log('File written successfully');return { message: 'File written successfully' };} catch (error) {console.log(error)throw new Error(`Error writing file: ${error.message}`);}}// 读取文件async readFile (path) {const url = `${this.baseUrl}${path}?op=OPEN&user.name=${this.user}`;try {const response = await axios.get(url, { responseType: 'stream' });return response.data;} catch (error) {throw new Error(`Error reading file: ${error.message}`);}}// 初始化当天目录async initDir () {const nowDayDir = getCurrentDateFormatted()this.originPagePdfPath = `${hdfsPagePdfPath}/${nowDayDir}`this.originAttachPath = `${hdfsAttachPath}/${nowDayDir}`// 创建页面快照目录await this.createDirectory(this.originPagePdfPath)// 创建存储附件目录await this.createDirectory(this.originAttachPath)}
}export default WebHDFSClient;

使用示例：

const hdfsClient = new WebHDFSClient()
// 读取文件内容
const fileData = fs.createReadStream(filePath);
// 写入文件到 hadoop
await hdfsClient.writeFile(saveAttachPath, fileData)