Text/String 当我在 Reducer 中读取它们时，Mapper 中发送的值是错误的

Question

我正在 Mapper 中发送一些数据，当我尝试在 Reducer 中读取它们时，它们发生了一些变化。在简历中，我使用 set 函数填充数据，然后使用 get 函数在 reducer 中读取它们。我不明白为什么如果我执行 println，数据会不同。

我发送的数据在一个名为 "ValorFechaHora" 的 class 中，有 3 个变量 Medicion、Fecha 和 Hora：

public class ValorFechaHora implements Writable {

 private IntWritable Medicion;
 private Text Fecha;
 private Text Hora;
 
 public void ValorFechaHora(){
  
 }
 
 public void ValorFechaHora(IntWritable Medicion, Text Fecha, Text Hora){
  setMedicion(Medicion);
  setFecha(Fecha);
  setHora(Hora);
 }
 
 public IntWritable getMedicion() {
  return Medicion;
 }

 public void setMedicion(IntWritable medicion) {
  Medicion = medicion;
 }

 public Text getFecha() {
  return Fecha;
 }

 public void setFecha(Text fecha) {
  Fecha = fecha;
 }

 public Text getHora() {
  return Hora;
 }

 public void setHora(Text hora) {
  Hora = hora;
 }

 @Override
 public int hashCode() {
  final int prime = 31;
  int result = 1;
  result = prime * result + ((Fecha == null) ? 0 : Fecha.hashCode());
  result = prime * result + ((Hora == null) ? 0 : Hora.hashCode());
  result = prime * result
    + ((Medicion == null) ? 0 : Medicion.hashCode());
  return result;
 }

 @Override
 public boolean equals(Object obj) {
  if (this == obj)
   return true;
  if (obj == null)
   return false;
  if (getClass() != obj.getClass())
   return false;
  ValorFechaHora other = (ValorFechaHora) obj;
  if (Fecha == null) {
   if (other.Fecha != null)
    return false;
  } else if (!Fecha.equals(other.Fecha))
   return false;
  if (Hora == null) {
   if (other.Hora != null)
    return false;
  } else if (!Hora.equals(other.Hora))
   return false;
  if (Medicion == null) {
   if (other.Medicion != null)
    return false;
  } else if (!Medicion.equals(other.Medicion))
   return false;
  return true;
 }

 public void readFields(DataInput in) throws IOException {
  
  Medicion = new IntWritable(in.readInt());
  Fecha = new Text(in.readLine());
  Hora = new Text(in.readLine());
 }
 
 public void write(DataOutput out) throws IOException {
  Medicion.write(out);
  Fecha.write(out);
  Hora.write(out);
 } 

}

在这里你可以看到我的映射器：

public static class LogsMapper extends
   Mapper<LongWritable, Text, Text, ValorFechaHora> {

  //En el mapper emitimos lo que leemos. Key = Dirección MAC. Value = Medición + Fecha + Hora
  
  private Text outKey = new Text();
  private ValorFechaHora outValue = new ValorFechaHora();
  
  @Override
  protected void map(LongWritable offset, Text line, Context context)
    throws IOException, InterruptedException {
   
   
   // Utilizamos row_auxiliar y row para leer los datos correctos (El offset no nos interesa)
   // Ejemplo de dato de entrada tras salir del filtrado básico "2536816 -47dB;8C:3A:E3:92:CB:3E;2014-11-12;14:22:20.795806"
   
   String row_auxiliar[] = line.toString().split("\t");
   String row[] = row_auxiliar[1].split(";");
   
   // Los datos en row quedan... ---> row[0]= Medicion row[1]= MAC row[2]= Fecha row[3]= Hora
 
   //Elegimos la MAC como key
   outKey = new Text(row[1]);
   
   //Elegimos la Medicion, Fecha y Hora como value
   outValue.setMedicion(new IntWritable(Integer.valueOf(row[0].substring(0,3))));
   outValue.setFecha(new Text(row[2]));
   outValue.setHora(new Text(row[3]));
   
   context.write(outKey, outValue);
    
  };

这是我的减速器：

public static class MaxReducer extends
   Reducer<Text, ValorFechaHora, Text, Text> {
  
  //En el reduce por ahora únicamente contamos el número de veces que ha sido la MAC registrada
  
  protected void reduce(Text MAC,
    Iterable<ValorFechaHora> values, Context context)
    throws IOException, InterruptedException {

   Text outKey = new Text();
   Text outValue = new Text();
   
   outKey = MAC;
   int sum = 0;
   
   for(ValorFechaHora val : values){
    System.out.println("1" + " " + val.getMedicion().toString());
    System.out.println("2" + " " + val.getFecha().toString());
    System.out.println("3" + " " + val.getHora().toString());
    
    sum = sum +1;
   }
   
   outValue = new Text(Integer.toString(sum));
   
   context.write(outKey, outValue);
  };

嗯，我不明白为什么当我做 bucle reducer 中的变量 val.getFecha().toString() 与变量 outKey.getFecha( ).toString 在映射器

蒂亚

Answer 1

我不太明白为什么会这样，但我解决了更改 class "ValorFechaHora"

中的这些代码行的问题

 public void readFields(DataInput in) throws IOException {
  
  Medicion = new IntWritable(in.readInt());
  //Fecha = new Text(in.readLine());
  //Hora = new Text(in.readLine());

        //Those two lines for these ones:

        Fecha = new Text(Text.readString(in));
  Hora = new Text(Text.readString(in));

天宇

Answer 2

您使用了错误的方法调用来填充文本对象。您应该使用 Text 对象的 readFields 方法。

目前您正在尝试通过接受字符串作为参数的构造函数来填充 Text 对象。您不能只使用 in.readLine 从 DataInput 对象读回 String，因为 Text 对象在没有终止换行符的情况下序列化到数据流。

要解决这个问题，您应该重新使用您应该初始化您的变量，然后只使用 readFields 方法（这可能会对您的代码产生其他影响，因为您当前没有使用对象重新使用模式（这比为每个 K/V 对象创建新对象更有效）：

private IntWritable Medicion = new IntWritable();
private Text Fecha = new Text();
private Text Hora = new Text();

public void readFields(DataInput in) {
    Medicion.readFields(in);
    Fecha.readFields(in);
    Hora.readFields(in);
}

否则，要保持您的代码不变（但效率较低），只需按如下方式更新 readFields 方法：

public void readFields(DataInput in) {
    Medicion = new Text();
    Medicion.readFields(in);

    Fecha = new Text();
    Fecha.readFields(in);

    Hora = new Text();
    Hora.readFields(in);
}

Text/String 当我在 Reducer 中读取它们时，Mapper 中发送的值是错误的

Text/String values sent in the Mapper are wrong when I read them in the Reducer

java

hadoop

mapreduce

reduce-reduce-conflict