Hadoop Map Reduce可根据累计能耗阅读查找最大小时能耗

waxmsbnn  于 2022-11-21  发布在  Hadoop
关注(0)|答案(1)|浏览(201)

确定所有日期内所有住房单元的最大每小时耗电量。您的处理应生成一个值。
我有这个样本数据

  1. LOG_ID HOUSE_ID CONDATE CONHOUR ENERGY_READING FLAG
  2. 3682572 16 2019-01-01 05:21:50 11143.735496 0
  3. 3682573 16 2019-01-01 05:22:00 11143.738274 0
  4. 3682574 16 2019-01-01 05:22:10 11143.741052 0
  5. 3682575 16 2019-01-01 05:22:20 11143.74383 0
  6. 3682576 16 2019-01-01 05:22:30 11143.746608 0
  7. 3682577 16 2019-01-01 05:22:40 11143.749386 0
  8. 3682578 16 2019-01-01 05:22:50 11143.752164 0
  9. 3682579 16 2019-01-01 05:23:00 11143.754942 0
  10. 3682580 16 2019-01-01 05:23:10 11143.75772 0
  11. 3682581 16 2019-01-01 05:23:20 11143.760498 0
  12. 3682582 16 2019-01-01 05:23:30 11143.763276 0
  13. 3682583 16 2019-01-01 05:23:40 11143.766054 0
  14. 3682584 16 2019-01-01 05:23:50 11143.768832 0

这是样本数据。其中能量消耗是累积的。所以在这里我想找到每天的最大小时消耗。
我试过了:Map器类

  1. import org.apache.hadoop.io.*;
  2. import org.apache.hadoop.mapreduce.Mapper;
  3. import java.io.IOException;
  4. public class EnergyMapper
  5. extends Mapper<LongWritable, Text, Text, FloatWritable> {
  6. @Override
  7. protected void map(LongWritable key, Text value, Mapper.Context context) throws IOException, InterruptedException {
  8. String[] values = value.toString().split("\\s+");
  9. String date;
  10. int houseId;
  11. String time;
  12. double energyReading;
  13. EnergyValues ev;
  14. try {
  15. date = values[2];
  16. houseId = Integer.parseInt(values[1]);
  17. time = values[3];
  18. energyReading = Double.parseDouble(values[4]);
  19. ev = new EnergyValues(houseId,time,energyReading,date);
  20. }
  21. catch (Exception e){
  22. date = "NA";
  23. ev = new EnergyValues();
  24. }
  25. context.write(new Text(date), ev);
  26. }
  27. }

异径管等级

  1. import org.apache.hadoop.io.Text;
  2. import org.apache.hadoop.mapreduce.Reducer;
  3. import java.text.ParseException;
  4. import java.text.SimpleDateFormat;
  5. import java.util.Calendar;
  6. import java.util.Date;
  7. import java.io.IOException;
  8. public class EnergyReducer extends Reducer<Text, EnergyValues, Text, EnergyValues> {
  9. @Override
  10. protected void reduce(Text key, Iterable<EnergyValues> sales, Context context)
  11. throws IOException, InterruptedException {
  12. String Data = "";
  13. // int i = 0;
  14. double Consumption = 0.0D;
  15. double minConsumption = 0.0D;
  16. double MaxConsumption = 0.0D;
  17. for (EnergyValues amount: sales) {
  18. //maxConsumption =
  19. try {
  20. minConsumption = getHourlyComp(key,sales,amount.getTime());
  21. } catch (ParseException e) {
  22. throw new RuntimeException(e);
  23. }
  24. Consumption = minConsumption - amount.getEnergyReading() ;
  25. System.out.println("outside function +++++++++++++++++++++++++");
  26. System.out.println(amount.getEnergyReading() );
  27. MaxConsumption = Math.max(MaxConsumption, Consumption);
  28. Data = amount.getHouseId()+","+ ","+ amount.getTime() + "," + amount.getEnergyReading();
  29. }
  30. context.write(new Text(String.valueOf(MaxConsumption)), new EnergyValues(Data));
  31. }
  32. public double getHourlyComp(Text key, Iterable<EnergyValues> sales, String time)throws ParseException{
  33. float maxConsumption = 0.0F;
  34. for (EnergyValues amount: sales) {
  35. // String hours = time.substring(0, 2);
  36. // if (hours != "23"){
  37. if (key.toString() == amount.getDate()) {
  38. String myTime = time;
  39. SimpleDateFormat df = new SimpleDateFormat("HH:MM:SS");
  40. Date d = df.parse(myTime);
  41. Calendar cal = Calendar.getInstance();
  42. cal.setTime(d);
  43. cal.add(Calendar.HOUR, 1);
  44. String newTime = df.format(cal.getTime());
  45. if (newTime == amount.getTime()) {
  46. System.out.println("Inside function ----------------------");
  47. System.out.println(amount.getEnergyReading() );
  48. return amount.getEnergyReading();
  49. }
  50. }
  51. // }
  52. }
  53. return 0;
  54. }
  55. }

能源价值类

  1. import org.apache.hadoop.io.Writable;
  2. import java.io.DataInput;
  3. import java.io.DataOutput;
  4. import java.io.IOException;
  5. public class EnergyValues implements Writable {
  6. int houseId;
  7. String time;
  8. double energyReading;
  9. String date;
  10. String data;
  11. public EnergyValues(){
  12. this.houseId = 0;
  13. this.time = "";
  14. this.energyReading =0.0;
  15. this.date = "";
  16. }
  17. public EnergyValues( String value2) {
  18. this.data = value2;
  19. }
  20. public EnergyValues(int val1, String val2, double val3,String val4) {
  21. this.houseId = val1;
  22. this.time = val2;
  23. this.energyReading = val3;
  24. this.date =val4;
  25. }
  26. public int getHouseId() {
  27. return houseId;
  28. }
  29. public String getTime() {return time; }
  30. public double getEnergyReading(){return energyReading;}
  31. public String getDate(){return date;}
  32. @Override
  33. public void write(DataOutput dataOutput) throws IOException {
  34. dataOutput.writeInt(houseId);
  35. dataOutput.writeChars(time);
  36. dataOutput.writeDouble(energyReading);
  37. dataOutput.writeChars(date);
  38. }
  39. @Override
  40. public void readFields(DataInput dataInput) throws IOException {
  41. houseId = dataInput.readInt();
  42. time = String.valueOf(dataInput.readChar());
  43. energyReading = dataInput.readDouble();
  44. date = String.valueOf(dataInput.readChar());
  45. }
  46. }

但是当我运行这个程序时,我得到的结果是

  1. 0.0 EnergyValues@f79a760
  2. 0.0 EnergyValues@14f5da2c
  3. 0.0 EnergyValues@239b0f9d
  4. 0.0 EnergyValues@619bfe29
  5. 0.0 EnergyValues@1eb6749b
  6. 0.0 EnergyValues@652a7737
  7. 0.0 EnergyValues@2bef51f2
  8. 0.0 EnergyValues@650eab8
vof42yt1

vof42yt11#

在“能量值类”类中,您需要覆盖该类的to string方法,在reducer中传递to_string,您传递的是对象的值而不是字符串

相关问题