0

我正在尝试使用 Apache Storm 处理 GeoHash 代码流。我正在使用这个库和 Apache Storm 0.9.3。可以在此处输入链接描述中找到 python 的 geohash 详细信息。

目前，我在一个 BOLT 类的执行方法中面临同步问题。我尝试使用单个粗体，这给了我正确的输出。但是当我从一个螺栓螺纹变为两个或更多时。输出搞砸了。

其中一个 BOLT 的代码片段（只有这个有问题）是：

public static int PRECISION=6;
private OutputCollector collector;
BufferedReader br;
String lastGeoHash="NONE";
HashMap<String,Integer> map;
HashMap<String,String[]> zcd;
TreeMap<Integer,String> counts=new TreeMap<Integer,String>();
public void prepare( Map conf, TopologyContext context, OutputCollector collector ) 
{
    String line="";
    this.collector = collector;
    map=new HashMap<String,Integer>();
    zcd=new HashMap<String,String[]>();
    try {
        br = new BufferedReader(new FileReader("/tmp/zip_code_database.csv"));
        int i=0;
        while ((line = br.readLine()) != null) {
            if(i==0){
                String columns[]=line.split(",");
                for(int j=0;j<columns.length;j++){
                    map.put(columns[j],j);
                }
            }else{
                String []split=line.split(",");
                zcd.put(split[map.get("\"zip\"")],new String[]{split[map.get("\"state\"")],split[map.get("\"primary_city\"")]});
            }
            i++;
        }
        br.close();
    //  System.out.println(zcd);
    } catch (FileNotFoundException e) {
        e.printStackTrace();
    } catch (IOException e) {
        e.printStackTrace();
    }
    System.out.println("Initialize");
    initializeTreeMapAsPerOurRequirement(counts);
}

public void execute( Tuple tuple ) 
{
    String completeFile = tuple.getStringByField("string");//So, this data is generated by Spout and it contains the complete shape file where each line is separated by a new line character i.e. "\n"
    String lines[]=completeFile.split("\t");
    String geohash=lines[0];
    int count=Integer.parseInt(lines[1]);
    String zip=lines[2];
    String best="";
    String city="";
    String state="";

    if(!(geohash.equals(lastGeoHash)) && !(lastGeoHash.equals("NONE"))){
        //if(counts.size()!=0){
            //System.out.println(counts.firstKey());
                best=counts.get(counts.lastKey());
                //System.out.println(geohash);
                if(zcd.containsKey("\""+best+"\"")){
                    city = zcd.get("\""+best+"\"")[0];
                    state = zcd.get("\""+best+"\"")[1];
                    System.out.println(lastGeoHash+","+best+","+state+","+city+","+"US");
                }else if(!best.equals("NONE")){
                    System.out.println(lastGeoHash);
                    city="MISSING";
                    state="MISSING";
                }
        //      initializeTreeMapAsPerOurRequirement(counts);
            //}else{
                //System.out.println("else"+geohash);
            //}

        //}
    }
    lastGeoHash=geohash;
    counts.put(count, zip);

    collector.ack( tuple );
}

private void initializeTreeMapAsPerOurRequirement(TreeMap<Integer,String> counts){
    counts.clear();
    counts.put(-1,"NONE");
}

public void declareOutputFields( OutputFieldsDeclarer declarer ) 
{
    System.out.println("here");
    declarer.declare( new Fields( "number" ) );
}

拓扑代码为：

public static void main(String[] args) 
{
    TopologyBuilder builder = new TopologyBuilder();

    builder.setSpout( "spout", new SendWholeFileDataSpout(),2);
    builder.setBolt( "map", new GeoHashBolt(),2).shuffleGrouping("spout");
    builder.setBolt("reduce",new GeoHashReduceBolt(),2).fieldsGrouping("map", new Fields("value"));

    Config conf = new Config();

    LocalCluster cluster = new LocalCluster();
    cluster.submitTopology("test", conf, builder.createTopology());
    Utils.sleep(10000);
    cluster.killTopology("test");
    cluster.shutdown();
}

有人可以查看代码并指导我一下。

4

1 回答 1

0

您已将parallelism_hint喷口和两个螺栓的 设置为 2。这意味着每个组件将运行 2 个执行器，这可能会弄乱您的输出。
通过设置parallelism_hint为 1，您可以获得所需的输出。

于 2015-02-24T20:13:59.370 回答