{"id":1313,"date":"2015-08-18T17:53:58","date_gmt":"2015-08-18T09:53:58","guid":{"rendered":"http:\/\/www.strongd.net\/?p=1313"},"modified":"2015-08-18T17:56:08","modified_gmt":"2015-08-18T09:56:08","slug":"hadoopdatatransport-hadoop%e6%95%b0%e6%8d%ae%e7%a7%bb%e5%8a%a8%e6%96%b9%e6%b3%95","status":"publish","type":"post","link":"https:\/\/www.strongd.net\/?p=1313","title":{"rendered":"HadoopDataTransport Hadoop\u6570\u636e\u79fb\u52a8\u65b9\u6cd5"},"content":{"rendered":"

\u524d\u6bb5\u65f6\u95f4\u65e7HADOOP\u5347\u7ea7\u5185\u5b58\uff0c\u9700\u8981\u628a\u67d0\u4e9b\u5171\u7528\u6570\u636e\u6587\u4ef6\u8fc1\u79fb\u5230\u4e34\u65f6\u7684HADOOP\u4e0a\uff0c\u5982\u679c\u7528hadoop \u547d\u4ee4\u6548\u7387N\u4f4e\uff0c
\n\u4e8e\u662f\u5c31\u5199\u4e00\u4e2a\u5c0f\u7a0b\u5e8f\uff0c\u6309\u7167\u6587\u4ef6\u5217\u8868\uff0c\u628a\u6570\u636e\u8fc1\u79fb\u5230\u65b0HADOOP\u4e0a\uff0c\u547d\u4ee4\uff1a<\/p>\n

java -classpath .\/lib\/hadoop-core-1.0.2.jar:.\/lib\/commons-logging-1.1.1.jar:.\/lib\/commons-configuration-1.6.jar:.\/lib\/commons-lang-2.4.jar:.\/ test.HadoopDataTransport filelist.txt 100<\/p>\n

[root@datanode4 ~]#\u00a0cat filelist.txt<\/p>\n

\/201401\/21\/3bb30e5f-cf3e-4182-a7c0-ed486f80a87a.mp3
\n\/201401\/21\/1d62fff3-744e-41c9-8152-5243ee0ce7b4.mp3
\n\/201401\/21\/784a53f4-2125-4bc6-bf6a-0adf40981a64.mp3<\/p>\n

\u4ee3\u7801\u6e05\u5355\uff1a\u4e0d\u559c\u6b22\u5199\u6ce8\u91ca\u3002<\/p>\n

package test;<\/p>\n

import java.io.BufferedReader;
\nimport java.io.FileReader;
\nimport java.io.FileWriter;
\nimport java.io.IOException;
\nimport java.io.InputStream;
\nimport java.io.PrintWriter;
\nimport java.util.ArrayList;
\nimport java.util.List;<\/p>\n

import org.apache.hadoop.conf.Configuration;
\nimport org.apache.hadoop.fs.FSDataInputStream;
\nimport org.apache.hadoop.fs.FSDataOutputStream;
\nimport org.apache.hadoop.fs.FileSystem;
\nimport org.apache.hadoop.fs.Path;
\nimport org.apache.hadoop.io.IOUtils;<\/p>\n

public class HadoopDataTransport implements Runnable {<\/p>\n

private static FileSystem src_fs;
\nprivate static FileSystem des_fs;
\nprivate static List<String> FailureList = new ArrayList<String>();<\/p>\n

private static List<Boolean> jobs = new ArrayList<Boolean>();<\/p>\n

static final String basedir = “\/data\/album”; \/\/ \u4fdd\u5b58\u5728HADOOP\u4e2d\u7684\u6839\u76ee\u5f55<\/p>\n

public static void main(String[] args) throws Exception {
\nList<String> al = new ArrayList<String>();
\nSystem.out.println(“read list from file : ” + args[0]);
\nint pm = Integer.parseInt(args[1]);
\nFileReader fr = new FileReader(args[0]);
\nBufferedReader br = new BufferedReader(fr);
\nString line;
\nwhile ((line = br.readLine()) != null) {
\nif (line!=null && line.length()>10) {
\nal.add(line);
\n}
\n}
\nint pmax = pm;
\nif(al.size()>0){
\nfor(int i=0;i<al.size();i=i+pmax){
\nint max = ((i+pmax)>al.size()?al.size():(i+pmax));
\nnew Thread(new HadoopDataTransport(al.subList(i, max),i,0)).start();
\njobs.add(true);
\n}
\n}
\nbr.close();
\nfr.close();
\nwhile(jobs.size()>0){
\nThread.sleep(500);
\n}
\nif(FailureList.size()>0){\/\/ save failure list
\nsaveFailurelist(args[0]);
\n}
\n}<\/p>\n

private List<String> filelist;
\nprivate int threadid;
\npublic HadoopDataTransport(List<String> list,int id,int opts){
\nthis.filelist = list;
\nthis.threadid = id;
\n}<\/p>\n

@Override
\npublic void run() {
\nint success = 0;
\nint failure =0;<\/p>\n

for(String file : filelist){
\nSystem.out.println(threadid + ” ==> processing …. “+file);
\ntry {
\nInputStream in = readFromHadoopInputStream(file);
\nif(in !=null){
\nString filename = createHdfsFile(file, in);
\nSystem.out.println(threadid + ” ==> “+filename + ” …… done.”);
\nsuccess++;
\n}
\n} catch (Exception e) {
\nAddFailure(file);
\nSystem.out.println(threadid + ” ==> “+file + ” …… failure.” + e.getMessage());
\nfailure++;
\n}
\n}
\nSystem.out.println(“===============ThreadId: “+threadid+” Finished Total:” + filelist.size() +” Success : “+ success+” Failure :”+ failure+”==========”);
\njobs.remove(true);
\n}<\/p>\n

private static void AddFailure(String filename){
\nFailureList.add(filename);
\n}
\nprivate static void saveFailurelist(String failurefile) throws Exception{
\nSystem.out.println(“Save “+”failure_”+failurefile);
\nFileWriter w = new FileWriter(“failure_”+failurefile);
\nPrintWriter out = new PrintWriter(w);
\nfor(String s : FailureList){
\nout.println(s);
\n}
\nout.close();
\n}<\/p>\n

private static String createHdfsFile(String dst, InputStream in)
\nthrows Exception {
\nFSDataOutputStream out =des_fs.create(
\ngetPath(dst));
\nIOUtils.copyBytes(in, out, 256 * 1024);
\nout.close();
\nreturn dst;
\n}<\/p>\n

private static InputStream readFromHadoopInputStream(String dst) throws Exception {
\nPath path = getPath(dst);
\nif (src_fs.exists(path)) {
\nFSDataInputStream is = src_fs.open(path);
\nreturn is;
\n} else {
\nthrow new Exception(“the file is not found .”);
\n}
\n}<\/p>\n

private static Path getPath(String dir) {
\nreturn new Path(basedir + dir);
\n}<\/p>\n

static {
\nConfiguration src_conf = new Configuration();
\nsrc_conf.set(“fs.default.name”, “hdfs:\/\/192.168.2.50:8020”); \/\/ conf.set(“fs.default.name”,
\n\/\/ “hdfs:\/\/namenode-backup-vip:8020”);
\nsrc_conf.set(“dfs.block.size”, “524288”);
\nsrc_conf.set(“dfs.replication”, “2”);
\nsrc_conf.set(“dfs.permissions”, “false”);
\nsrc_conf.set(“dfs.permissions.supergroup”, “resin”);
\nsrc_conf.set(“dfs.web.ugi”, “resin”);
\ntry {
\nsrc_fs = FileSystem.get(src_conf);
\n} catch (IOException e) {
\ne.printStackTrace();
\n}
\nSystem.out
\n.println(“Initialize Hadoop Server src fs hdfs:\/\/192.168.2.50:8020”);<\/p>\n

Configuration des_conf = new Configuration();
\ndes_conf.set(“fs.default.name”, “hdfs:\/\/192.168.2.85:8020”); \/\/ conf.set(“fs.default.name”,
\n\/\/ “hdfs:\/\/namenode-backup-vip:8020”);
\ndes_conf.set(“dfs.block.size”, “524288”);
\ndes_conf.set(“dfs.replication”, “2”);
\ndes_conf.set(“dfs.permissions”, “false”);
\ndes_conf.set(“dfs.permissions.supergroup”, “resin”);
\ndes_conf.set(“dfs.web.ugi”, “resin”);
\ntry {
\ndes_fs = FileSystem.get(des_conf);
\n} catch (IOException e) {
\ne.printStackTrace();
\n}
\nSystem.out
\n.println(“Initialize Hadoop Server des fs hdfs:\/\/192.168.2.85:8020”);
\n}<\/p>\n

}<\/p><\/blockquote>\n","protected":false},"excerpt":{"rendered":"

\u524d\u6bb5\u65f6\u95f4\u65e7HADOOP\u5347\u7ea7\u5185\u5b58\uff0c\u9700\u8981\u628a\u67d0\u4e9b\u5171\u7528\u6570\u636e\u6587\u4ef6\u8fc1\u79fb\u5230\u4e34\u65f6\u7684HADOOP\u4e0a\uff0c\u5982\u679c\u7528hadoop \u547d\u4ee4\u6548\u7387N\u4f4e\uff0c \u4e8e\u662f\u5c31\u5199\u4e00\u4e2a\u5c0f\u7a0b\u5e8f\uff0c\u6309\u7167\u6587\u4ef6\u5217\u8868\uff0c\u628a\u6570\u636e\u8fc1\u79fb\u5230\u65b0HADOOP\u4e0a\uff0c\u547d\u4ee4\uff1a java -classpath .\/lib\/hadoop-core-1.0.2.jar:.\/lib\/commons-logging-1.1.1.jar:.\/lib\/commons-configuration-1.6.jar:.\/lib\/commons-lang-2.4.jar:.\/ test.HadoopDataTransport filelist.txt 100 [root@datanode4 ~]#\u00a0cat filelist.txt \/201401\/21\/3bb30e5f-cf3e-4182-a7c0-ed486f80a87a.mp3 \/201401\/21\/1d62fff3-744e-41c9-8152-5243ee0ce7b4.mp3 \/201401\/21\/784a53f4-2125-4bc6-bf6a-0adf40981a64.mp3 \u4ee3\u7801\u6e05\u5355\uff1a\u4e0d\u559c\u6b22\u5199\u6ce8\u91ca\u3002 package test; import java.io.BufferedReader; import java.io.FileReader; import java.io.FileWriter; import java.io.IOException; import java.io.InputStream; import java.io.PrintWriter; import java.util.ArrayList; import java.util.List; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FSDataInputStream; import org.apache.hadoop.fs.FSDataOutputStream; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.IOUtils; public class HadoopDataTransport implements Runnable { private static FileSystem … Continue reading HadoopDataTransport Hadoop\u6570\u636e\u79fb\u52a8\u65b9\u6cd5<\/span><\/a><\/p>\n","protected":false},"author":2,"featured_media":0,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[1],"tags":[],"class_list":["post-1313","post","type-post","status-publish","format-standard","hentry","category-java"],"_links":{"self":[{"href":"https:\/\/www.strongd.net\/index.php?rest_route=\/wp\/v2\/posts\/1313","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/www.strongd.net\/index.php?rest_route=\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/www.strongd.net\/index.php?rest_route=\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/www.strongd.net\/index.php?rest_route=\/wp\/v2\/users\/2"}],"replies":[{"embeddable":true,"href":"https:\/\/www.strongd.net\/index.php?rest_route=%2Fwp%2Fv2%2Fcomments&post=1313"}],"version-history":[{"count":2,"href":"https:\/\/www.strongd.net\/index.php?rest_route=\/wp\/v2\/posts\/1313\/revisions"}],"predecessor-version":[{"id":1315,"href":"https:\/\/www.strongd.net\/index.php?rest_route=\/wp\/v2\/posts\/1313\/revisions\/1315"}],"wp:attachment":[{"href":"https:\/\/www.strongd.net\/index.php?rest_route=%2Fwp%2Fv2%2Fmedia&parent=1313"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/www.strongd.net\/index.php?rest_route=%2Fwp%2Fv2%2Fcategories&post=1313"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/www.strongd.net\/index.php?rest_route=%2Fwp%2Fv2%2Ftags&post=1313"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}