重慶分公司,新征程啟航
為企業提供網站建設、域名注冊、服務器等服務
為企業提供網站建設、域名注冊、服務器等服務
這篇文章給大家介紹如何實現一個MapReduce讀取數據存入HBase,內容非常詳細,感興趣的小伙伴們可以參考借鑒,希望對大家能有所幫助。
成都創新互聯公司專業為企業提供襄城網站建設、襄城做網站、襄城網站設計、襄城網站制作等企業網站建設、網頁設計與制作、襄城企業網站模板建站服務,十多年襄城做網站經驗,不只是建網站,更提供有價值的思路和整體網絡服務。
車輛位置數據文件,格式:車輛id 速度:油耗:當前里程。
通過MapReduce算出每輛車的平均速度、油耗、里程
vid1 78:8:120 vid1 56:11:124 vid1 98:5:130 vid1 72:6:131 vid2 78:4:281 vid2 58:9:298 vid2 67:15:309
創建Map類和map函數
import java.io.IOException; import java.util.StringTokenizer; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Mapper; public class VehicleMapper extends Mapper
創建Reduce類
import java.io.IOException; import org.apache.hadoop.hbase.client.Put; import org.apache.hadoop.hbase.io.ImmutableBytesWritable; import org.apache.hadoop.hbase.mapreduce.TableReducer; import org.apache.hadoop.hbase.util.Bytes; import org.apache.hadoop.io.Text; public class VehicleReduce extends TableReducer{ @Override public void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException { int speed = 0; int oil = 0; int mile = 0; int count = 0; for (Text val : values) { String str = val.toString(); String[] arr = str.split(":"); speed += Integer.valueOf(arr[0]); oil += Integer.valueOf(arr[1]); mile += Integer.valueOf(arr[2]) - mile; // 累積里程 count++; } speed = (int) speed / count; // 求平均值 oil = (int) oil / count; mile = (int) mile / count; String result = speed + ":" + oil + ":" + mile; Put put = new Put(key.getBytes()); put.add(Bytes.toBytes("info"), Bytes.toBytes("property"), Bytes.toBytes(result)); ImmutableBytesWritable keys = new ImmutableBytesWritable(key.getBytes()); context.write(keys, put); } }
運行任務
import java.io.IOException; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; import org.apache.hadoop.hbase.HBaseConfiguration; import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; public class VehicleMapReduceJob { public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException { Configuration conf = new Configuration(); conf = HBaseConfiguration.create(conf); Job job = new Job(conf, "HBase_VehicleInfo"); job.setJarByClass(VehicleMapReduceJob.class); job.setMapperClass(VehicleMapper.class); job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(Text.class); FileInputFormat.addInputPath(job, new Path(args[0])); // 設置輸入文件路徑 TableMapReduceUtil.initTableReducerJob("vehicle", VehicleReduce.class, job); System.exit(job.waitForCompletion(true) ? 0 : 1); } }
將代碼導出成vehicle.jar,放在hadoop-1.2.1目錄下,輸入命令
./bin/hadoop jar vehicle.jar com/xh/vehicle/VehicleMapReduceJob input/vehicle.txt
HBase結果查詢:
關于如何實現一個MapReduce讀取數據存入HBase就分享到這里了,希望以上內容可以對大家有一定的幫助,可以學到更多知識。如果覺得文章不錯,可以把它分享出去讓更多的人看到。