Changes between Initial Version and Version 1 of ProductEmp

Timestamp:

Jan 16, 2012, 9:10:52 PM (12 years ago)

Author:

waue

Comment:

--

ProductEmp

@@ +1 @@
+{{{
+#!java
+
+
+package datatable;
+
+/*
+ * Info :
+ *              http://trac.nchc.org.tw/cloud/wiki/waue/2012/0117
+ */
+
+import java.io.BufferedReader;
+import java.io.BufferedWriter;
+import java.io.File;
+import java.io.FileReader;
+import java.io.FileWriter;
+import java.io.IOException;
+import java.util.HashMap;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.IntWritable;
+import org.apache.hadoop.io.LongWritable;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.mapreduce.Job;
+import org.apache.hadoop.mapreduce.Mapper;
+import org.apache.hadoop.mapreduce.Reducer;
+import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
+import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
+
+public class ProductEmp {
+
+        public static class DtMap extends
+                        Mapper<LongWritable, Text, Text, IntWritable> {
+                private IntWritable one = new IntWritable(1);
+
+                public void map(LongWritable key, Text value, Context context)
+                                throws IOException, InterruptedException {
+                        String s[] = value.toString().trim().split(":");
+                        String str = s[1];
+                        context.write(new Text(str), one);
+                }
+        }
+
+        public static class DtReduce extends Reducer<Text, IntWritable, Text, Text> {
+
+                public HashMap<String, Integer> hm_product_map = new HashMap<String, Integer>();
+                public HashMap<String, String> hm_product_name_map = new HashMap<String, String>();
+                private Configuration conf;
+
+                
+                protected void setup(Context context) throws IOException,
+                                InterruptedException {
+                        // put the product list into HashMap for querying
+                        conf = context.getConfiguration();
+
+                        String str = conf.get("dt_product_list");
+                        // 取得到 編號1=價格1=名稱1, 編號2=價格2=名稱2...
+                        String[] arr = str.split(",");
+                        String[] tmp_arr;
+                        for (String tmp : arr) {
+                                tmp_arr = tmp.split("=");
+                                hm_product_map.put(tmp_arr[0], new Integer(tmp_arr[1]));
+                                hm_product_name_map.put(tmp_arr[0], tmp_arr[2]);
+                        }
+                        System.err.println(hm_product_map.toString());
+                        System.err.println(hm_product_name_map.toString());
+                }
+
+                public void reduce(Text key, Iterable<IntWritable> values,
+                                Context context) throws IOException, InterruptedException {
+                        // <key,value> 為 <編號, 串列(111..1)>
+
+                        int count = 0, price = 0, sales = 0;
+                        // 取得每個 key 的(111..1)計數，相加，得到 數量
+                        for (IntWritable val : values) {
+                                count += val.get();
+                        }
+                        // 取得編號的單價
+                        String id = key.toString().trim();
+                        String p_name = "";
+                        //// System.err.println("id=" + id + " <e> ");
+                        if (id != null) {
+                                price = (hm_product_map.get(id) == null) ? 0 : hm_product_map
+                                                .get(id);
+                                p_name = hm_product_name_map.get(id);
+
+                        }
+                        //// System.err.println("price=" + price + " <e> ");
+                        //// System.err.println("name=" + p_name + " <e> ");
+
+                        // 銷售額 = 數量 X 單價
+                        if (count != 0 && price != 0)
+                                sales = count * price;
+
+                        // 輸出的value 設定為 銷售額
+
+                        Text t_info = new Text("[\'" + id + "\',\'" + p_name + "\',");
+                        Text t_result = new Text("\'" + price + "\',\'" + count + "\',\'"
+                                        + sales + "\'],");
+                        // 輸出新的 <key,value> 
+                        context.write(t_info, t_result);
+                }
+        }
+
+        static public boolean checkAndDelete(String out, Configuration conf) {
+                Path dstPath = new Path(out);
+                try {
+                        FileSystem hdfs = dstPath.getFileSystem(conf);
+                        if (hdfs.exists(dstPath)) {
+                                hdfs.delete(dstPath, true);
+                        }
+
+                } catch (IOException e) {
+                        e.printStackTrace();
+                        return false;
+                }
+                return true;
+        }
+
+        static boolean putToHdfs(String src, String dst, Configuration conf) {
+                Path dstPath = new Path(dst);
+                try {
+                        // 產生操作hdfs的物件
+                        FileSystem hdfs = dstPath.getFileSystem(conf);
+                        // 上傳
+                        hdfs.copyFromLocalFile(false, new Path(src), new Path(dst));
+
+                } catch (IOException e) {
+                        e.printStackTrace();
+                        return false;
+                }
+                return true;
+        }
+
+        static public boolean deleteFolder(File dir) {
+                File filelist[] = dir.listFiles();
+                int listlen = filelist.length;
+                for (int i = 0; i < listlen; i++) {
+                        if (filelist[i].isDirectory()) {
+                                deleteFolder(filelist[i]);
+                        } else {
+                                filelist[i].delete();
+                        }
+                }
+                return dir.delete();
+        }
+
+        public static boolean download_dt_js(String from, String dst,
+                        Configuration conf) {
+
+                Path hdfs_result_path = new Path(from);
+                Path local_result_path = new Path(dst);
+                try {
+                        FileSystem hdfs = hdfs_result_path.getFileSystem(conf);
+                        if (hdfs.exists(hdfs_result_path)) {
+                                hdfs.copyFromLocalFile(hdfs_result_path, local_result_path);
+
+                        }
+
+                } catch (IOException e) {
+                        e.printStackTrace();
+                        return false;
+                }
+                return true;
+
+        }
+
+        public static boolean transfer_dt_js(String from, String dst) {
+
+                String b = null;
+                String header = "var aDataSet = [";
+                String footer = "];";
+                File file = new File(dst);
+                BufferedReader br;
+                BufferedWriter bw;
+                if (!file.exists() != false) {
+                        try {
+                                file.createNewFile();
+
+                        } catch (IOException e) {
+                                e.printStackTrace();
+                        }
+                }
+                try {
+                        bw = new BufferedWriter(new FileWriter(file));
+                        bw.write(header);
+                        FileReader fr = new FileReader(from);
+                        br = new BufferedReader(fr);
+                        while ((b = br.readLine()) != null) {
+                                //// System.out.println(b);
+                                bw.write(b);
+                                bw.newLine();
+                                bw.flush();
+                        }
+                        bw.write(footer);
+                        bw.close();
+                        br.close();
+                } catch (Exception e) {
+
+                        System.err.println(e.getMessage());
+                        return false;
+                }
+                return true;
+        }
+
+        public static void main(String args[]) throws Exception {
+
+                // (0). 設定系統環境參數
+                String local_store_product_list = "/tmp/ProductEmp/conf/store.txt";
+                String local_input_dir = "/tmp/ProductEmp/input";
+                String hdfs_input_dir = "/tmp/ProductEmp/hdfs";
+                String hdfs_output_dir = "/tmp/ProductEmp/out";
+                String local_download_out_file = "/tmp/ProductEmp/result/tmp.log";
+                String local_db_file = "/tmp/ProductEmp/www-web/db.js";
+
+                /*
+                 * // 參數設定功能，暫不開放 String[] argc = { local_input_dir, hdfs_output_dir,
+                 * local_store_product_list }; args = argc;
+                 * 
+                 * if (args.length != 3) { System.err
+                 * .println("Usage: hadoop jar ProductEmp.jar <input> <output> <store.txt>"
+                 * ); System.exit(2); }
+                 */
+
+                // (1). 設定hadoop 參數
+                Configuration conf = new Configuration();
+
+                // 1.1 設定成單機測試模式
+                conf.set("mapred.job.tracker", "local"); // for single
+                conf.set("fs.default.name", "file:///"); // for single
+
+                // (2) 取得商品資訊
+                // 2.1 : 0.編號 1.名稱 2.價格
+                String str, product_list = "";
+                String[] arr;
+                BufferedReader bfr = new BufferedReader(new FileReader(new File(
+                                local_store_product_list)));
+                while ((str = bfr.readLine()) != null) {
+                        arr = str.split(";");
+
+                        // 2.2 編號=價格=名稱
+                        product_list += arr[0].trim() + "=" + arr[2].trim() + "="
+                                        + arr[1].trim() + ",";
+                }
+                System.out.println("您設定的商品清單:" + product_list);
+                conf.set("dt_product_list", product_list);
+
+                // (3) hdfs 操作
+                // 3.1 檢查hdfs 上的結果目錄是否存在，已經在的話刪除之
+                if (checkAndDelete(hdfs_output_dir, conf))
+                        System.out.println("已刪除 舊的輸出結果 :" + hdfs_output_dir);
+                if (checkAndDelete(hdfs_input_dir, conf))
+                        System.out.println("已刪除 舊的輸入資料 :" + hdfs_input_dir);
+
+                // 3.2 將 local input 送上 hdfs
+                if (putToHdfs(local_input_dir, hdfs_input_dir, conf))
+                        System.out.println("已將 新的輸入資料 送上 hdfs :" + local_input_dir + " => "
+                                        + hdfs_input_dir);
+
+                // (4) hadoop 運算設定
+                Job job = new Job(conf, "Product Calculate");
+                job.setJarByClass(ProductEmp.class);
+                job.setMapperClass(DtMap.class);
+                job.setReducerClass(DtReduce.class);
+                job.setOutputKeyClass(Text.class);
+                job.setOutputValueClass(IntWritable.class);
+                FileInputFormat.addInputPath(job, new Path(hdfs_input_dir));
+                FileOutputFormat.setOutputPath(job, new Path(hdfs_output_dir));
+                // 4.n 送出運算指令
+                if ( job.waitForCompletion(true) )
+                        System.out.println("hadoop 運算完成");
+                
+                // (5) 下載
+                if ( download_dt_js(hdfs_output_dir + "/part-r-00000",
+                                local_download_out_file, conf) ){
+                        System.out.println("從 hdfs 下載到 local 完成");
+                }else{
+                        System.err.println("[失敗] 從 hdfs 下載到 local 失敗");
+                }
+                        
+                // (6) 改寫
+                if (  transfer_dt_js(local_download_out_file, local_db_file)  ){
+                        System.out.println("轉換資料格式完成，請將" + local_db_file + "放到 網頁伺服器");
+                }else{
+                        System.err.println("[失敗] 轉換成database失敗");
+                }
+
+        }
+}
+
+
+
+
+
+
+
+}}}