热门标签 | HotTags
当前位置:  开发笔记 > 运维 > 正文

hadoop上传文件功能实例代码

这篇文章主要介绍了hadoop上传文件功能实例代码,需要的朋友可以参考下

hdfs上的文件是手动执行命令从本地linux上传至hdfs的。在真实的运行环境中,我们不可能每次手动执行命令上传的,这样太过繁琐。那么,我们可以使用hdfs提供的Java api实现文件上传至hdfs,或者直接从ftp上传至hdfs。 

然而,需要说明一点,之前笔者是要运行MR,都需要每次手动执行yarn jar,在实际的环境中也不可能每次手动执行。像我们公司是使用了索答的调度平台/任务监控平台,可以定时的以工作流执行我们的程序,包括普通java程序和MR。其实,这个调度平台就是使用了quartz。当然,这个调度平台也提供其它的一些功能,比如web展示、日志查看等,所以也不是免费的。 

首先,给大家简单介绍一下hdfs。hdfs是以流式数据访问模式来存储超大文件,hdfs的构建思路是一次写入,多次读取,这样才是最高效的访问模式。hdfs是为高数据吞吐量应用优化的,所以会以提高时间延迟为代价。对于低延时的访问需求,我们可以使用hbase。 

然后,还要知道hdfs中块(block)的概念,默认为64MB。块是hdfs的数据读写的最小单位,通常每个map任务一次只处理一个block,像我们对集群性能评估就会使用到这个概念,比如目前有多少节点,每个节点的磁盘空间、cpu以及所要处理的数据量、网络带宽,通过这些信息来进行性能评估。我们可以使用Hadoop fsck / -files -blocks列出文件系统中各个文件由哪些块构成。 

然后,再就是要知道namenode和datanode,这个在之前的博文已经介绍过,下面看看cm环境中hdfs的管理者(namenode)和工作者(datanode),如下 

在yarn环境中是可以有多个nameNode的。此环境中没有SecondaryNameNode,当然也可以有。 

好了,关于hdfs的基本概念就讲到这儿了,下面来看看具体的代码。

一、java实现上传本地文件至hdfs

这里,可以直接使用hdfs提供的java api即可实现,代码如下:

package com.bjpowernode.hdfs.local;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
/**
 * ClassName:UploadLocalFileToHdfs 
* Function: 本地文件上传至hdfs.
* Date: 2016年3月28日 下午10:06:05
* @author qiyongkang * @version * @since JDK 1.6 * @see */ public class UploadLocalFileToHdfs { public static void main(String[] args) { Configuration cOnf= new Configuration(); String localDir = "/home/qiyongkang"; String hdfsDir = "/qiyongkang"; try{ Path localPath = new Path(localDir); Path hdfsPath = new Path(hdfsDir); FileSystem hdfs = FileSystem.get(conf); hdfs.copyFromLocalFile(localPath, hdfsPath); }catch(Exception e){ e.printStackTrace(); } } }

注意,这里hdfs上传目录如果不存在的话,hdfs会自动创建,比较智能。 

打完包后,上传至服务器,执行yarn jar mr-demo-0.0.1-SNAPSHOT-jar-with-dependencies.jar,然后执行hadoop fs -ls /qiyongkang便可看到: 

二、java实现上传ftp上的文件至hdfs

首先,我们得准备一个ftp服务器,关于ftp服务器的搭建,大家可以查阅资料,笔者就不赘述了。 

其实,从ftp上拉取文件上传到hdfs上,这个过程大家不要想复杂了,我们讲本地文件上传到hdfs,其实就是采用流的方式。因此,我们可以直接读取ftp上的文件流,然后以流的方式写入到hdfs。 

下面,直接贴出代码:

package com.bjpowernode.hdfs.ftp;
import java.io.InputStream;
import org.apache.commons.net.ftp.FTP;
import org.apache.commons.net.ftp.FTPClient;
import org.apache.commons.net.ftp.FTPFile;
import org.apache.commons.net.ftp.FTPReply;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
/**
 * ClassName:UploadFtpFileToHdfs 
* Function: TODO ADD FUNCTION.
* Reason: TODO ADD REASON.
* Date: 2016年3月28日 下午10:50:37
* * @author qiyongkang * @version * @since JDK 1.6 * @see */ public class UploadFtpFileToHdfs { public static void main(String[] args) { Configuration cOnf= new Configuration(); loadFromFtpToHdfs("172.31.26.200", "qiyongkang", "qyk123456", "/www/input/", "/qiyongkang/", conf); } /** * * loadFromFtpToHdfs:将数据从ftp上传到hdfs上.
* * @author qiyongkang * @param ip * @param username * @param password * @param filePath * @param outputPath * @param conf * @return * @since JDK 1.6 */ private static boolean loadFromFtpToHdfs(String ip, String username, String password, String filePath, String outputPath, Configuration conf) { FTPClient ftp = new FTPClient(); InputStream inputStream = null; FSDataOutputStream outputStream = null; boolean flag = true; try { ftp.connect(ip); ftp.login(username, password); ftp.setFileType(FTP.BINARY_FILE_TYPE); ftp.setControlEncoding("UTF-8"); int reply = ftp.getReplyCode(); if (!FTPReply.isPositiveCompletion(reply)) { ftp.disconnect(); } FTPFile[] files = ftp.listFiles(filePath); FileSystem hdfs = FileSystem.get(conf); for (FTPFile file : files) { if (!(file.getName().equals(".") || file.getName().equals(".."))) { inputStream = ftp.retrieveFileStream(filePath + file.getName()); outputStream = hdfs.create(new Path(outputPath + file.getName())); IOUtils.copyBytes(inputStream, outputStream, conf, false); if (inputStream != null) { inputStream.close(); ftp.completePendingCommand(); } } } ftp.disconnect(); } catch (Exception e) { flag = false; e.printStackTrace(); } return flag; } }

然后同样打包上传后执行yarn jar mr-demo-0.0.1-SNAPSHOT-jar-with-dependencies.jar,便可看到: 

总结

以上所述是小编给大家介绍的hadoop上传文件功能实例代码,希望对大家有所帮助,如果大家有任何疑问请给我留言,小编会及时回复大家的。在此也非常感谢大家对网站的支持!


推荐阅读
  • 在搭建Hadoop集群以处理大规模数据存储和频繁读取需求的过程中,经常会遇到各种配置难题。本文总结了作者在实际部署中遇到的典型问题,并提供了详细的解决方案,帮助读者避免常见的配置陷阱。通过这些经验分享,希望读者能够更加顺利地完成Hadoop集群的搭建和配置。 ... [详细]
  • 本文介绍了如何使用Flume从Linux文件系统收集日志并存储到HDFS,然后通过MapReduce清洗数据,使用Hive进行数据分析,并最终通过Sqoop将结果导出到MySQL数据库。 ... [详细]
  • 深入解析:存储技术的演变与发展
    本文探讨了从单机文件系统到分布式文件系统的存储技术发展过程,详细解释了各种存储模型及其特点。 ... [详细]
  • Hadoop 2.6 主要由 HDFS 和 YARN 两大部分组成,其中 YARN 包含了运行在 ResourceManager 的 JVM 中的组件以及在 NodeManager 中运行的部分。本文深入探讨了 Hadoop 2.6 日志文件的解析方法,并详细介绍了 MapReduce 日志管理的最佳实践,旨在帮助用户更好地理解和优化日志处理流程,提高系统运维效率。 ... [详细]
  • 构建高可用性Spark分布式集群:大数据环境下的最佳实践
    在构建高可用性的Spark分布式集群过程中,确保所有节点之间的无密码登录是至关重要的一步。通过在每个节点上生成SSH密钥对(使用 `ssh-keygen -t rsa` 命令并保持默认设置),可以实现这一目标。此外,还需将生成的公钥分发到所有节点的 `~/.ssh/authorized_keys` 文件中,以确保节点间的无缝通信。为了进一步提升集群的稳定性和性能,建议采用负载均衡和故障恢复机制,并定期进行系统监控和维护。 ... [详细]
  • 从理想主义者的内心深处萌发的技术信仰,推动了云原生技术在全球范围内的快速发展。本文将带你深入了解阿里巴巴在开源领域的贡献与成就。 ... [详细]
  • 精选10款Python框架助力并行与分布式机器学习
    随着神经网络模型的不断深化和复杂化,训练这些模型变得愈发具有挑战性,不仅需要处理大量的权重,还必须克服内存限制等问题。本文将介绍10款优秀的Python框架,帮助开发者高效地实现分布式和并行化的深度学习模型训练。 ... [详细]
  • 如何在U8系统中连接服务器并获取数据
    本文介绍了如何在U8系统中通过不同的方法连接服务器并获取数据,包括使用MySQL客户端连接实例的方法,如非SSL连接和SSL连接,并提供了详细的步骤和注意事项。 ... [详细]
  • 本文详细介绍了 `org.apache.hadoop.hdfs.server.namenode.FSNamesystem.shouldUseDelegationTokens()` 方法的用途和实际应用场景,并提供了多个代码示例以帮助开发者更好地理解和使用该方法。 ... [详细]
  • Ubuntu 环境下配置 LAMP 服务器
    本文详细介绍了如何在 Ubuntu 系统上安装和配置 LAMP(Linux、Apache、MySQL 和 PHP)服务器。包括 Apache 的安装、PHP 的配置以及 MySQL 数据库的设置,确保读者能够顺利搭建完整的 Web 开发环境。 ... [详细]
  • Hadoop Datanode DataXceiver 错误处理问题
    Ambari 每分钟会向 Datanode 发送一次“ping”请求以确保其正常运行。然而,Datanode 在处理空内容时没有相应的逻辑,导致出现错误。 ... [详细]
  • 大数据领域的职业路径与角色解析
    本文将深入探讨大数据领域的各种职业和工作角色,帮助读者全面了解大数据行业的需求、市场趋势,以及从入门到高级专业人士的职业发展路径。文章还将详细介绍不同公司对大数据人才的需求,并解析各岗位的具体职责、所需技能和经验。 ... [详细]
  • Hadoop的文件操作位于包org.apache.hadoop.fs里面,能够进行新建、删除、修改等操作。比较重要的几个类:(1)Configurati ... [详细]
  • Python 数据可视化实战指南
    本文详细介绍如何使用 Python 进行数据可视化,涵盖从环境搭建到具体实例的全过程。 ... [详细]
  • 本文探讨了 Kafka 集群的高效部署与优化策略。首先介绍了 Kafka 的下载与安装步骤,包括从官方网站获取最新版本的压缩包并进行解压。随后详细讨论了集群配置的最佳实践,涵盖节点选择、网络优化和性能调优等方面,旨在提升系统的稳定性和处理能力。此外,还提供了常见的故障排查方法和监控方案,帮助运维人员更好地管理和维护 Kafka 集群。 ... [详细]
author-avatar
lmc的
这个家伙很懒,什么也没留下!
PHP1.CN | 中国最专业的PHP中文社区 | DevBox开发工具箱 | json解析格式化 |PHP资讯 | PHP教程 | 数据库技术 | 服务器技术 | 前端开发技术 | PHP框架 | 开发工具 | 在线工具
Copyright © 1998 - 2020 PHP1.CN. All Rights Reserved | 京公网安备 11010802041100号 | 京ICP备19059560号-4 | PHP1.CN 第一PHP社区 版权所有