@@ -0,0 +1,2368 @@
+package edu.neu.ccs.pyramid.elasticsearch;%0A%0Aimport java.io.Serializable;%0Aimport java.util.*;%0Aimport java.util.concurrent.ConcurrentHashMap;%0Aimport java.util.stream.IntStream;%0A%0A/**%0A * Created by chengli on 1/20/15.%0A */%0Apublic class DuplicateDetector implements Serializable%7B%0A    private static final long serialVersionUID = 1L;%0A    private transient ESIndex esIndex;%0A    Set%3CString%3E allDuplicates;%0A%0A    public DuplicateDetector(ESIndex esIndex) %7B%0A        this.esIndex = esIndex;%0A        this.allDuplicates = Collections.newSetFromMap(new ConcurrentHashMap%3CString, Boolean%3E());%0A    %7D%0A%0A    public void addDuplicates(String id1, String id2)%7B%0A        allDuplicates.add(id1);%0A        allDuplicates.add(id2);%0A    %7D%0A%0A    public void detect()%7B%0A        Map%3CInteger, Set%3CString%3E%3E lengthToIds = new ConcurrentHashMap%3C%3E();%0A        int numDocs = esIndex.getNumDocs();%0A        IntStream.range(0,numDocs).parallel()%0A                .forEach(i -%3E %7B%0A                    Map%3CInteger,String%3E termVector = esIndex.getTermVectorFromIndex(%22%22+i);%0A                    if (!lengthToIds.containsKey(termVector.size()))%7B%0A                        lengthToIds.put(termVector.size(),Collections.newSetFromMap(new ConcurrentHashMap%3CString, Boolean%3E()));%0A                    %7D%0A                    lengthToIds.get(termVector.size()).add(%22%22+i);%0A                %7D);%0A        lengthToIds.entrySet().stream().parallel().map(Map.Entry::getValue)%0A                .forEach(this::check);%0A%0A    %7D%0A%0A    private void check(Set%3CString%3E candidates)%7B%0A        List%3CString%3E list = new ArrayList%3C%3E(candidates);%0A%0A%0A        int size = candidates.size();%0A        if (size==1)%7B%0A            return;%0A        %7D%0A        for (int i=0;i%3Csize-1;i++)%7B%0A            for (int j=i+i;j%3Csize;j++)%7B%0A                String id1 = list.get(i);%0A                String id2 = list.get(j);%0A                Map%3CInteger,String%3E termVector1 = esIndex.getTermVectorFromIndex(id1);%0A                Map%3CInteger,String%3E termVector2 = esIndex.getTermVectorFromIndex(id2);%0A                if (termVector1.equals(termVector2))%7B%0A                    addDuplicates(id1,id2);%0A                %7D%0A            %7D%0A        %7D%0A    %7D%0A%0A    @Override%0A    public String toString() %7B%0A        StringBuilder sb = new StringBuilder();%0A        for (String str: allDuplicates)%7B%0A            sb.append(str);%0A            sb.append(%22,%22);%0A        %7D%0A        return sb.toString();%0A    %7D%0A%7D%0A

add class DuplicateDetector