通过HA方式操作HDFS
2024-09-07 16:06:56
之前操作hdfs的时候，都是固定namenode的地址，然后去操作。这个时候就必须判断namenode的状态为active还是standby，比较繁琐，如果集群使用了HA的形式，就很方便了
直接上代码，看注释：
package com.ideal.template.openbigdata.util; import java.io.IOException;
import java.net.URI;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.sql.Timestamp;
import java.text.SimpleDateFormat; import java.util.LinkedList;
import java.util.List; //import org.anarres.lzo.LzoAlgorithm;
//import org.anarres.lzo.LzoDecompressor;
//import org.anarres.lzo.LzoInputStream;
//import org.anarres.lzo.LzoLibrary;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.log4j.Logger; public class HadoopUse
{
private static Log log = LogFactory.getLog(HadoopUse.class); /**
* 设置hdfs配置信息
* @return
*/
private static Configuration getConf()
{
Configuration conf = new Configuration(); //设置配置相关的信息，分别对应hdfs-site.xml core-site.xml
conf.set("fs.defaultFS", "hdfs://dragoncluster");
conf.set("dfs.nameservices", "dragoncluster");
conf.set("dfs.ha.namenodes.dragoncluster", "nn1,nn2");
conf.set("dfs.namenode.rpc-address.dragoncluster.nn1", "n01.dragon.com:8020");
conf.set("dfs.namenode.rpc-address.dragoncluster.nn2", "n02.dragon.com:8020");
conf.set("dfs.client.failover.proxy.provider.dragoncluster", "org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider"); //设置实现类，因为会出现类覆盖的问题
conf.set("fs.hdfs.impl", org.apache.hadoop.hdfs.DistributedFileSystem.class.getName());
conf.set("fs.file.impl", org.apache.hadoop.fs.LocalFileSystem.class.getName());
return conf;
} /**
* 设置kerberos认证
* @param conf
* @throws Exception
*/
private static void kerberosLogin(Configuration conf) throws Exception
{
conf.set("hadoop.security.authentication", "kerberos");
UserGroupInformation.setConfiguration(conf);
UserGroupInformation.loginUserFromKeytab("openbigdata@DRAGON.COM", "/etc/security/keytabs/openbigdata.keytab");
} public static long getSize(String uri, String user)
{
Path path = new Path(URI.create(uri)); Configuration conf = new Configuration();
try
{
FileSystem fs = FileSystem.get(URI.create(uri), conf, user);
return fs.getContentSummary(path).getLength() / 1024 / 1024; // 单位为MB
}
catch (Exception ex)
{
log.error("HadoopUse.getSize" + ex.getMessage(), ex);
return 0;
}
} /**
* 在hdfs上创建文件，并写入内容
*
* @param uri
* @param content
* @param user
* @return
*/
public static boolean createHdfsFile(String uri, String user, String fullName, String content)
{
if (fullName == null || fullName.length() == 0)
{// 本地路径不正确
return false;
}
if (content == null || content.length() == 0)
{// hdfs路径不正确
return false;
} try
{
Configuration conf = new Configuration(); FileSystem fs = FileSystem.get(URI.create(uri), conf, user);
FSDataOutputStream os = null; if (fs.exists(new Path(fullName)) == true)
{// 如果该路径存在
// os = fs.append(new Path(fullName));
fs.delete(new Path(fullName), true);
}
os = fs.create(new Path(fullName));
os.write(content.getBytes());
os.close();
fs.close();
return true;
}
catch (Exception ex)
{
log.error("HadoopUse.createHdfsFile" + ex.getMessage(), ex);
return false;
}
} /**
* 删除hdfs上的文件
* @param uri
* @param user
* @param fullName
* @return
*/
public static boolean deleteHdfsFile(String uri, String user, String fullName)
{
if (fullName == null || fullName.length() == 0)
{// 本地路径不正确
log.error("HadoopUse.deleteHdfsFile文件名不合法");
return false;
} try
{
Configuration conf = new Configuration(); FileSystem fs = FileSystem.get(URI.create(uri), conf, user);
//FSDataOutputStream os = null; if (fs.exists(new Path(fullName)) == true)
{// 如果该路径存在
// os = fs.append(new Path(fullName));
fs.delete(new Path(fullName), true);
}
return true;
}
catch (Exception ex)
{
log.error("HadoopUse.createHdfsFile" + ex.getMessage(), ex);
}
return false;
} /**
* 根据resultset将值写入到hdfs上
* @param uri
* @param user
* @param fullName
* @param resultSet
* @param terminated
* @return
* @throws InterruptedException
* @throws IOException
* @throws SQLException
*/
public void createHdfsFile(String fullName, ResultSet resultSet, String terminated, FlagUtil flag)
throws IOException, InterruptedException, SQLException, Exception
{
if (resultSet == null)
{ // 如果查询出来的游标为空，直接退出
return;
}
SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); FileSystem fs = null;
FSDataOutputStream out = null;
Configuration conf = getConf();
kerberosLogin(conf); fs = FileSystem.get(conf);
if (fs.exists(new Path(fullName)) == true)
{// 如果该路径存在
fs.delete(new Path(fullName), true);
} // 获取文件句柄
out = fs.create(new Path(fullName)); // 写入文件内容
ResultSetMetaData rsmd = resultSet.getMetaData();
int rowCnt = rsmd.getColumnCount();
int count = 0;
while (resultSet.next())
{
count++;
if(count >= 1000)
{//每1000条记录检查一次需要终止任务
if(flag.getTeminalStatus() == true)
{
break;
}
count = 0;
} for (int i = 1; i <= rowCnt; i++)
{
if (resultSet.getObject(i) == null)
{// 如果是空的数据
out.write("".getBytes("utf-8"));
}
else
{
String item = null;
if("DATE".equals(rsmd.getColumnTypeName(i).toUpperCase()))
{//如果是日期类型
Timestamp date = resultSet.getTimestamp(i);
item = sdf.format(date);
}
else
{
item = String.valueOf(resultSet.getObject(i));
}
if (item != null)
{
out.write(item.getBytes("utf-8"));
}
else
{
out.write("".getBytes("utf-8"));
}
}
if (i < rowCnt)
{// 如果写完一列，则插入分隔符
out.write(terminated.getBytes("utf-8"));
}
}
// 切换到下一行
out.write("\r\n".getBytes("utf-8"));
}
log.info("fullName:" + fullName + "写入成功"); if (out != null)
{
out.flush();
out.close();
}
if (fs != null)
{
fs.close();
}
} /**
* 查询路径
* @param path
* @return
* @throws Exception
*/
public static List<String> listDir(String path) throws Exception
{
Configuration conf = getConf();
kerberosLogin(conf);
FileSystem fs = FileSystem.get(conf); Path hdfs = new Path(path);
List<String> pathList = null;
FileStatus files[] = fs.listStatus(hdfs);
if(files!=null && files.length >0)
{
pathList = new LinkedList<String>();
for (FileStatus file : files)
{
pathList.add(file.getPath().toString());
}
}
return pathList;
} public static void main(String[] args) throws Exception
{
List<String> pathList = listDir(args[0]);
for(String path: pathList)
{
System.out.println(path);
}
}
}
注意，这用到了HA，以及kerberos认证，
最新文章
- Html特殊字符表
- 统计SqlServer每张表内的数据量
- ASP.NET MVC分页组件MvcPager 2.0版发布暨网站全新改版
- EPANET源码中用到的几个简单C语言函数介绍三
- oracle-5-的升级步骤
- codeforces 719B：Anatoly and Cockroaches
- 清除数据库中大于10W行的垃圾历史数据
- C++ 常见崩溃问题分析
- iOS自定义NavigationBar
- 【2016北京集训测试赛】river
- python 判断变量是否是 None 的三种写法
- numpy教程：统计函数Statistics
- kolla 多节点部署 openstack
- [Linux]目录x权限对文件操作的影响
- 如何在疲劳的JS世界中持续学习
- 显示AVI文件的桢数
- 按条件设置gridcontrol 单元格属性
- Java实验案例（接口）
- spring-data-jpa快速入门（一）——整合阿里Druid
- MySQL 安装 + Windows7
热门文章
- [原创]java实现word转pdf
- 理解分布式id生成算法SnowFlake
- BZOJ_2253_[2010 Beijing wc]纸箱堆叠 _CDQ分治+树状数组
- BZOJ_4327_JSOI2012 玄武密码_AC自动机
- maven(二)创建一个maven的web项目中解决Cannot change version of project facet Dynamic web module to 2.5
- C#面向对象之数据库（理论、插入、修改、删除、查询）
- NYOJ1——A+B Problem NYOJ2——括号配对问题
- STL - merge()
- （水题）洛谷 - P1051 - 谁拿了最多奖学金
- 2016多校8th 1008【线段树-神题】