java调用API操作HDFS
本篇文章介绍Java调用API从hdfs读取数据
package mongodb;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.URI;
import java.util.Arrays;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.io.compress.CompressionCodecFactory;
import org.apache.hadoop.io.compress.CompressionInputStream;
import org.apache.hadoop.util.ReflectionUtils;
class Item implements Comparable
{
String value;
double weight;
public Item(String v)
{
value=v;
weight = Double.parseDouble(value.split(":")[1]);
}
public int compareTo(Object o)
{
return this.weight == ((Item) o).weight ? 0 :
(this.weight > ((Item) o).weight ? -1 : 1);
}
}
public class BatchUpdateSim {
public static String parse(String str)
{
//String str="90003 20718001:1.0,2077635:1.0,2053809:1.0";
String[] fields=str.split("\t");
String[] valueArray= fields[1].split(",");
Item[] items = new Item[valueArray.length];
for(int i=0;i<items.length;i++)
{
items[i]=new Item(valueArray[i]);
}
Arrays.sort(items);
for(int i=0;i<valueArray.length;i++)
{
valueArray[i] = "{"+items[i].value+"}";
}
String valueStr = StringUtils.join(valueArray,",");
return "{\"key\":"+fields[0]+",\"values\":["+valueStr+"]}";
}
public static void main(String[] args) throws IOException, ClassNotFoundException {
// TODO Auto-generated method stub
Configuration conf = new Configuration ();
conf.addResource("/usr/local/hadoop/conf/core-site.xml");
conf.addResource("/usr/local/hadoop/conf/hdfs-site.xml");
conf.addResource("/usr/local/hadoop/conf/mapred-site.xml");
//String HDFS="hdfs://webdm-cluster";
//String HDFS="hdfs://localhost:9000";
String HDFS=args[0];
FileSystem hdfs = FileSystem.get(URI.create(HDFS),conf);
String filePath=args[1];
FSDataInputStream fin = hdfs.open(new Path(filePath));
//CompressionCodecFactory factory = new CompressionCodecFactory(conf);
//CompressionCodec codec = factory.getCodec(new Path(filePath)); //依据hdfs文件的后缀类型主动辨认
//Class<?> codecClass = Class.forName("com.hadoop.compression.lzo.LzoCodec");
//CompressionCodec codec = (CompressionCodec)ReflectionUtils.newInstance(codecClass, conf);
BufferedReader reader = null;
String line;
int count=0;
RecsysDb db = RecsysDb.getInstance();
String itemSimName=args[2];
try
{
// if (codec == null)
// {
reader = new BufferedReader(new InputStreamReader(fin)); // in = new BufferedReader(new InputStreamReader(fin, "UTF-8"));
// }
/*
else
{
System.out.println("辨认出紧缩类型");
CompressionInputStream comInputStream = codec.createInputStream(fin);
reader = new BufferedReader(new InputStreamReader(comInputStream));
} */
while ((line = reader.readLine()) != null)
{
//if(count==0) System.out.println(line);
String strJson=parse(line);
db.updateItemSim(itemSimName, strJson);
count++;
if(count%1000==0)System.out.println("count:"+count);
}
}
finally
{
if (reader != null)
{
reader.close();
}
System.out.println(count);
}
}
}
packag