Tôi đang sử dụng Hadoop để phân tích dữ liệu GSOD (ftp://ftp.ncdc.noaa.gov/pub/data/gsod/). Tôi chọn 5 năm để thực hiện thí nghiệm của mình (2005 - 2009). Tôi đã định cấu hình một cụm nhỏ và thực hiện chương trình MapReduce đơn giản để truy xuất các bản ghi nhiệt độ cao nhất trong một năm.

Bây giờ tôi phải tạo một chương trình MR mới để tính tất cả các hiện tượng đã xảy ra trong nhiều năm ở mỗi địa điểm.

File tôi phải phân tích có cấu trúc như sau:

STN---...FRSHTO
722115 110001
722115 011001
722110 111000
722110 001000
722000 001000

Cột STN thể hiện mã địa điểm, FRSHTT thể hiện các hiện tượng: F - sương mù, R - mưa hoặc mưa phùn, S - tuyết hoặc viên băng, H - mưa đá, T - sấm sét, O - lốc xoáy hoặc đám mây hình phễu.

Giá trị 1 có nghĩa là hiện tượng đó xảy ra vào ngày đó; giá trị 0 có nghĩa là hiện tượng đó không xảy ra.

Tôi cần tìm kết quả như:

722115: F = 1, R = 2, S = 1, O = 2
722110: F = 1, R = 1, S = 2
722000: S = 1

Tôi có thể chạy chương trình MR, nhưng kết quả sai, cho tôi kết quả như sau:

722115 F, 1
722115 R, 1
722115 R, 1
722115 S, 1
722115 Ô, 1
722115 Ô, 1
722110 F, 1
722110 R, 1
722110 S, 1
722110 S, 1
722000 S, 1

Tôi đã sử dụng các mã này:

người lập bản đồ.java

lớp công khai Mapper mở rộng org.apache.hadoop.mapreduce.Mapper {
@Ghi đè
bản đồ khoảng trống được bảo vệ (khóa LongWritable, giá trị văn bản, bối cảnh org.apache.hadoop.mapreduce.Mapper.Context) ném IOException, InterruptedException {
    Dòng chuỗi = value.toString();
    // Mỗi file đều bắt đầu bằng một dòng mô tả trường nên tôi bỏ qua dòng này
    if (!line.startsWith("STN---")) {
        // Trường đầu tiên của dòng có nghĩa là mã trạm nơi dữ liệu được thu thập
        Trạm chuỗi = line.substring(0, 6);
        Chuỗi sương mù = (line.substring(132, 133));
        Chuỗi rainOrDrizzle = (line.substring(133, 134));
        Chuỗi snowOrIcePellets = (line.substring(134, 135));
        Chuỗi mưa đá = (line.substring(135, 136));
        Chuỗi sấm = (line.substring(136, 137));
        Chuỗi lốc xoáyOrFunnelCloud = (line.substring(137, 138));

        if (fog.equals("1"))
            context.write(Station mớiPhenomenun(station,"F"), IntWritable mới(1));
        if (rainOrDrizzle.equals("1"))
            context.write(Station mớiPhenomenun(station,"R"), IntWritable(1) mới);
        if (snowOrIcePellets.equals("1"))
            context.write(Station mớiPhenomenun(station,"S"), IntWritable mới(1));
        if (mưa đá.equals("1"))
            context.write(Station mớiPhenomenun(station,"H"), IntWritable(1) mới);
        if (thunder.equals("1"))
            context.write(Station mớiPhenomenun(station,"T"), IntWritable(1) mới);
        if (tornadoOrFunnelCloud.equals("1"))
            context.write(Station mớiPhenomenun(station,"O"), IntWritable mới(1));
    }
}
}

Giảm tốc.java

lớp công khai Trình giảm tốc mở rộng org.apache.hadoop.mapreduce.Reducer {

protected void less(StationPhenomenun key, Iterable value, org.apache.hadoop.mapreduce.Reducer.Context context) ném IOException, InterruptedException {
số int = 0;        
    for (Giá trị IntWritable: giá trị) {
        count++;
    }

    Trạm chuỗi = key.getStation().toString();
    Lần xuất hiện chuỗi = key.getPhenomenun().toString();

    StationHiện tượng textPair = new StationPhenomenun(trạm, lần xuất hiện);
    context.write(textPair, new IntWritable(count));
}
}

StationPhenomenum.java

lớp công khai StationPhenomenun triển khai WritableComparable {
trạm dây riêng;
hiện tượng chuỗi riêng tư;
public StationPhenomenun(Trạm chuỗi, Hiện tượng chuỗi) {
    this.station = trạm;
    this.phenomenun = hiện tượng;
}
trạm công cộngPhenomenun() {
}
Chuỗi công khai getStation() {
    trạm trở về;
}
Chuỗi công khai getPhenomenun() {
    hiện tượng quay trở lại;
}
@Ghi đè
public void readFields(DataInput in) ném IOException {
    trạm = in.readUTF();
    hiện tượng = in.readUTF();
}
@Ghi đè
public void write(DataOutput out) ném IOException {
    out.writeUTF(trạm);
    out.writeUTF(hiện tượng);
}
@Ghi đè
public int so sánhTo(StationPhenomenun t) {
    int cmp = this.station.compareTo(t.station);
    nếu (cmp != 0) {
        trả lại cmp;
    }
    trả lại this.phenomenun.compareTo(t.phenomenun);
}    
@Ghi đè
boolean công khai bằng(Object obj) {
    nếu (obj == null) {
        trả về sai;
    }
    if (getClass() != obj.getClass()) {
        trả về sai;
    }
    Final StationPhenomenun other = (StationPhenomenun) obj;
    if (this.station != other.station && (this.station == null || !this.station.equals(other.station))) {
        trả về sai;
    }
    if (this.phenomenun != other.phenomenun && (this.phenomenun == null || !this.phenomenun.equals(other.phenomenun))) {
        trả về sai;
    }
    trả về đúng sự thật;
}
@Ghi đè
public int hashCode() {
    trả về this.station.hashCode() * 163 + this.phenomenun.hashCode();
}
}

NcdcJob.java

lớp công khai NcdcJob {
public static void main(String[] args) throws Exception {
    Cấu hình conf = Cấu hình mới();
    Công việc công việc = Công việc mới (conf);
    job.setJarByClass(NcdcJob.class);
    FileInputFormat.addInputPath(job, new Path("/user/hadoop/input"));
    FileOutputFormat.setOutputPath(job, new Path("/user/hadoop/station"));
    job.setMapperClass(Mapper.class);
    job.setReducerClass(Reducer.class);
    job.setMapOutputKeyClass(StationPhenomenun.class);
    job.setMapOutputValueClass(IntWritable.class);
    job.setOutputKeyClass(StationPhenomenun.class);
    job.setOutputValueClass(IntWritable.class);
    System.exit(job.waitForCompletion(true) ? 0 : 1);
}
}

Có ai đã làm điều gì tương tự chưa?

Tái bút: Tôi đã thử giải pháp này ( Hadoop - khóa tổng hợp ) nhưng không hiệu quả với tôi.

câu trả lời hay nhất

Chỉ cần kiểm tra xem 2 lớp sau có khớp với cách triển khai tùy chỉnh của bạn hay không.

 job.setMapperClass(Mapper.class);
 job.setReducerClass(Reducer.class);

Tôi đã có thể nhận được kết quả mong muốn với những thay đổi sau

bản đồ void được bảo vệ (Khóa LongWritable, Giá trị văn bản, Ngữ cảnh) ném IOException, InterruptedException {

protected void less(StationPhenomenun key, Iterable value, Ngữ cảnh ngữ cảnh) ném IOException, InterruptedException {

Đồng thời đổi tên lớp thành MyMapper Và MyReducer

722115,1,1,0,0,0,1
722115,0,1,1,0,0,1
722110,1,1,1,0,0,0
722110,0,0,1,0,0,0
722000,0,0,1,0,0,0

Đối với bộ đầu vào này, tôi có thể nhận được kết quả sau

TrạmHiện tượng [trạm=722000, hiện tượng=S] 1
TrạmHiện tượng [trạm=722110, hiện tượng=F] 1
TrạmHiện tượng [trạm=722110, hiện tượng=R] 1
TrạmHiện tượng [trạm=722110, hiện tượng=S] 2
TrạmHiện tượng [trạm=722115, hiện tượng=F] 1
TrạmHiện tượng [trạm=722115, hiện tượng=O] 2
TrạmHiện tượng [trạm=722115, hiện tượng=R] 2
TrạmHiện tượng [trạm=722115, hiện tượng=S] 1

Cách tính cũng như vậy, bạn chỉ cần tùy chỉnh cách hiển thị đầu ra là được.

Về những khó khăn của Hadoop với khóa tổng hợp, chúng tôi đã tìm thấy một câu hỏi tương tự trên Stack Overflow: https://stackoverflow.com/questions/18381684/