backtype.storm.Config.setMaxSpoutPending()方法的使用及代码示例

x33g5p2x  于2022-01-18 转载在 其他  
字(11.4k)|赞(0)|评价(0)|浏览(90)

本文整理了Java中backtype.storm.Config.setMaxSpoutPending()方法的一些代码示例，展示了Config.setMaxSpoutPending()的具体用法。这些代码示例主要来源于Github/Stackoverflow/Maven等平台，是从一些精选项目中提取出来的代码，具有较强的参考意义，能在一定程度帮忙到你。Config.setMaxSpoutPending()方法的具体详情如下：
包路径：backtype.storm.Config
类名称：Config
方法名：setMaxSpoutPending

Config.setMaxSpoutPending介绍

暂无

代码示例

代码示例来源：origin: alibaba/jstorm

public void setMaxSpoutPending(int max) {
  setMaxSpoutPending(this, max);
}

代码示例来源：origin: alibaba/jstorm

public static void test() {
  
  conf.setMaxSpoutPending(20);
  
  String[] className = Thread.currentThread().getStackTrace()[1].getClassName().split("\\.");
  String topologyName = className[className.length - 1];
  
  if (isLocal) {
    drpc = new LocalDRPC();
  }
  
  try {
    JStormHelper.runTopology(buildTopology(drpc), topologyName, conf, 60, new JStormHelper.CheckAckedFail(conf),
        isLocal);
  } catch (Exception e) {
    // TODO Auto-generated catch block
    e.printStackTrace();
    Assert.fail("Failed");
  }
}

代码示例来源：origin: alibaba/jstorm

public static void test() {
  TopologyBuilder builder = new TopologyBuilder();
  
  builder.setSpout("spout", new InOrderSpout(), 8);
  builder.setBolt("count", new Check(), 8).fieldsGrouping("spout", new Fields("c1"));
  
  conf.setMaxSpoutPending(20);
  
  String[] className = Thread.currentThread().getStackTrace()[1].getClassName().split("\\.");
  String topologyName = className[className.length - 1];
  
  if (isLocal) {
    drpc = new LocalDRPC();
  }
  
  try {
    JStormHelper.runTopology(buildDevicesTopology(), topologyName, conf, 60,
        new JStormHelper.CheckAckedFail(conf), isLocal);
  } catch (Exception e) {
    // TODO Auto-generated catch block
    e.printStackTrace();
    Assert.fail("Failed");
  }
}

代码示例来源：origin: alibaba/jstorm

public static void test() {
  TopologyBuilder builder = new TopologyBuilder();
  
  builder.setSpout("spout", new InOrderSpout(), 8);
  builder.setBolt("count", new Check(), 8).fieldsGrouping("spout", new Fields("c1"));
  
  conf.setMaxSpoutPending(20);
  
  String[] className = Thread.currentThread().getStackTrace()[1].getClassName().split("\\.");
  String topologyName = className[className.length - 1];
  
  if (isLocal) {
    drpc = new LocalDRPC();
  }
  
  try {
    JStormHelper.runTopology(buildVehiclesTopology(), topologyName, conf, 60,
        new JStormHelper.CheckAckedFail(conf), isLocal);
  } catch (Exception e) {
    // TODO Auto-generated catch block
    e.printStackTrace();
    Assert.fail("Failed");
  }
}

代码示例来源：origin: alibaba/jstorm

public static void test() {
  TopologyBuilder builder = new TopologyBuilder();
  
  builder.setSpout("spout", new InOrderSpout(), 8);
  builder.setBolt("count", new Check(), 8).fieldsGrouping("spout", new Fields("c1"));
  
  conf.setMaxSpoutPending(20);
  
  String[] className = Thread.currentThread().getStackTrace()[1].getClassName().split("\\.");
  String topologyName = className[className.length - 1];
  
  if (isLocal) {
    drpc = new LocalDRPC();
  }
  
  try {
    JStormHelper.runTopology(buildTopology(drpc), topologyName, conf, 60, new DrpcValidator(), isLocal);
  } catch (Exception e) {
    // TODO Auto-generated catch block
    e.printStackTrace();
    Assert.fail("Failed");
  }
}

代码示例来源：origin: alibaba/jstorm

public static void test() {
  TopologyBuilder builder = new TopologyBuilder();
  
  builder.setSpout("spout", new InOrderSpout(), 8);
  builder.setBolt("count", new Check(), 8).fieldsGrouping("spout", new Fields("c1"));
  
  Config conf = new Config();
  conf.setMaxSpoutPending(20);
  
  String[] className = Thread.currentThread().getStackTrace()[1].getClassName().split("\\.");
  String topologyName = className[className.length - 1];
  
  if (isLocal) {
    drpc = new LocalDRPC();
  }
  
  try {
    JStormHelper.runTopology(buildTopology(drpc), topologyName, conf, 60, new DrpcValidator(), isLocal);
  } catch (Exception e) {
    // TODO Auto-generated catch block
    e.printStackTrace();
    Assert.fail("Failed");
  }
}

代码示例来源：origin: alibaba/jstorm

public static void test() {
  MemoryTransactionalSpout spout = new MemoryTransactionalSpout(DATA, new Fields("word"),
      PARTITION_TAKE_PER_BATCH);
  TransactionalTopologyBuilder builder = new TransactionalTopologyBuilder("global-count", "spout", spout, 1);
  builder.setBolt("partial-count", new BatchCount(), 2).noneGrouping("spout");
  builder.setBolt("sum", new UpdateGlobalCount(), 1).globalGrouping("partial-count");
  
  conf.setDebug(true);
  conf.setMaxSpoutPending(3);
  
  String[] className = Thread.currentThread().getStackTrace()[1].getClassName().split("\\.");
  String topologyName = className[className.length - 1];
  try {
    JStormHelper.runTopology(builder.buildTopology(), topologyName, conf, 60,
        new JStormHelper.CheckAckedFail(conf), isLocal);
  } catch (Exception e) {
    // TODO Auto-generated catch block
    e.printStackTrace();
    Assert.fail("Failed");
  }
}

代码示例来源：origin: com.n3twork.storm/storm-core

public void setMaxSpoutPending(int max) {
  setMaxSpoutPending(this, max);
}

代码示例来源：origin: com.alibaba.jstorm/jstorm-core

public void setMaxSpoutPending(int max) {
  setMaxSpoutPending(this, max);
}

代码示例来源：origin: com.twitter.heron/heron-storm

public void setMaxSpoutPending(int max) {
 setMaxSpoutPending(this, max);
}

代码示例来源：origin: pereferrera/trident-lambda-splout

public static void main(String[] args) throws Exception {
    Config conf = new Config();
    conf.setMaxSpoutPending(20);

    // This topology can only be run as local because it is a toy example
    LocalDRPC drpc = new LocalDRPC();
    LocalCluster cluster = new LocalCluster();
    cluster.submitTopology("hashtagCounter", conf, buildTopology(drpc));
    // Query 100 times for hashtag "california" for illustrating the effect of the lambda architecture
    for(int i = 0; i < 100; i++) {
      System.out.println("Result for hashtag 'california' -> " + drpc.execute("hashtags", "california"));
      Thread.sleep(1000);
    }
  }
}

代码示例来源：origin: mvogiatzis/first-stories-twitter

private static Config createTopologyConfiguration(Properties prop,
    boolean localMode) {
  Config conf = new Config();
  List<String> dprcServers = new ArrayList<String>();
  dprcServers.add("localhost");
  conf.put(Config.DRPC_SERVERS, dprcServers);
  conf.put(Config.DRPC_PORT, 3772);
  if (!localMode)
    conf.put(Config.STORM_CLUSTER_MODE, new String("distributed"));
  conf.put("UNIQUE_WORDS_EXPECTED",
      prop.getProperty("UNIQUE_WORDS_EXPECTED"));
  conf.put("PATH_TO_OOV_FILE", prop.getProperty("PATH_TO_OOV_FILE"));
  conf.put("L", prop.getProperty("L"));
  conf.put("BucketsParallelism", prop.getProperty("BucketsParallelism"));
  conf.put("k", prop.getProperty("k"));
  conf.put("QUEUE_SIZE", prop.getProperty("QUEUE_SIZE"));
  List<String> countAggKeepFields = new ArrayList<String>();
  countAggKeepFields.add("tweet_obj");
  countAggKeepFields.add("coltweet_obj");
  conf.put("countAggKeepFields", countAggKeepFields);
  conf.put("THRESHOLD", prop.getProperty("THRESHOLD"));
  conf.put("RECENT_TWEETS_TO_COMPARE_WITH",
      prop.getProperty("RECENT_TWEETS_TO_COMPARE_WITH"));
  conf.setDebug(false);
  conf.setNumWorkers(Integer.valueOf((String) prop
      .get("NUMBER_OF_WORKERS")));
  conf.setMaxSpoutPending(50000000);
  return conf;
}

代码示例来源：origin: openimaj/openimaj

@Override
public Config prepareConfig() {
  if (preparedConfig == null) {
    preparedConfig = new Config();
    preparedConfig.setMaxSpoutPending(500);
    preparedConfig.setNumWorkers(numberOfWorkers);
    preparedConfig.setFallBackOnJavaSerialization(false);
    preparedConfig.setSkipMissingKryoRegistrations(false);
    JenaStormUtils.registerSerializers(preparedConfig);
  }
  return preparedConfig;
}

代码示例来源：origin: qidasheng/storm-kafka-xlog

config.setNumWorkers(1);
config.setMaxTaskParallelism(10);
config.setMaxSpoutPending(10000);
config.put(Config.NIMBUS_HOST, nimbusIp);
config.put(Config.NIMBUS_THRIFT_PORT, 6627);

代码示例来源：origin: pranab/chombo

/**
 * @param topologyName
 * @param conf
 * @param builder
 * @throws AlreadyAliveException
 * @throws InvalidTopologyException
 */
public static void submitStormTopology(String topologyName, Config conf,  TopologyBuilder builder) 
    throws AlreadyAliveException, InvalidTopologyException {
  int numWorkers = ConfigUtility.getInt(conf, "num.workers", 1);
  int maxSpoutPending = ConfigUtility.getInt(conf, "max.spout.pending", 1000);
  int maxTaskParalleism = ConfigUtility.getInt(conf, "max.task.parallelism", 100);
  conf.setNumWorkers(numWorkers);
  conf.setMaxSpoutPending(maxSpoutPending);
  conf.setMaxTaskParallelism(maxTaskParalleism);
  StormSubmitter.submitTopology(topologyName, conf, builder.createTopology());
}

代码示例来源：origin: jrkinley/storm-hbase

/**
 * @param args
 * @throws InterruptedException 
 */
public static void main(String[] args) throws InterruptedException {
 // Add transactional spout
 MemoryTransactionalSpout spout = new MemoryTransactionalSpout(values,
   new Fields("shortid", "url", "user", "date"), 3);
 TransactionalTopologyBuilder builder = new TransactionalTopologyBuilder(
   "shorturl-count", "spout", spout, 2);
 // Build TupleTableConifg
 TupleTableConfig ttConfig = new TupleTableConfig("shorturl", "shortid");
 ttConfig.setBatch(false);
 ttConfig.addColumn("data", "clicks");
 ttConfig.addColumn("daily", "date");
 builder.setBolt("hbase-counters", new HBaseCountersBatchBolt(ttConfig), 2)
   .fieldsGrouping("spout", new Fields("shortid"));
 LocalCluster cluster = new LocalCluster();
 Config stormConfig = new Config();
 stormConfig.setDebug(true);
 stormConfig.setMaxSpoutPending(3);
 cluster.submitTopology("hbase-example", stormConfig,
   builder.buildTopology());
 Thread.sleep(10000);
 cluster.shutdown();
}

代码示例来源：origin: openimaj/openimaj

public static void main(String[] args) {
    final Config conf = new Config();
    conf.setDebug(false);
    conf.setNumWorkers(2);
    conf.setMaxSpoutPending(1);
    conf.setFallBackOnJavaSerialization(false);
    conf.setSkipMissingKryoRegistrations(false);
    final LocalCluster cluster = new LocalCluster();
    final TopologyBuilder builder = new TopologyBuilder();
    builder.setSpout("randomSpout1", new RandomFieldSpout(2, 0, 0, 1)); // (nfields,seed,min,max)
    builder.setSpout("randomSpout2", new RandomFieldSpout(2, 10, 0, 1)); // (nfields,seed,min,max)
    JoinBolt.connectNewBolt(builder);
    final StormTopology topology = builder.createTopology();
    cluster.submitTopology("playTopology", conf, topology);
    Utils.sleep(10000);
    cluster.killTopology("playTopology");
    cluster.shutdown();

  }
}

代码示例来源：origin: org.openimaj.storm/core-storm

public static void main(String[] args) {
    final Config conf = new Config();
    conf.setDebug(false);
    conf.setNumWorkers(2);
    conf.setMaxSpoutPending(1);
    conf.setFallBackOnJavaSerialization(false);
    conf.setSkipMissingKryoRegistrations(false);
    final LocalCluster cluster = new LocalCluster();
    final TopologyBuilder builder = new TopologyBuilder();
    builder.setSpout("randomSpout1", new RandomFieldSpout(2, 0, 0, 1)); // (nfields,seed,min,max)
    builder.setSpout("randomSpout2", new RandomFieldSpout(2, 10, 0, 1)); // (nfields,seed,min,max)
    JoinBolt.connectNewBolt(builder);
    final StormTopology topology = builder.createTopology();
    cluster.submitTopology("playTopology", conf, topology);
    Utils.sleep(10000);
    cluster.killTopology("playTopology");
    cluster.shutdown();

  }
}

代码示例来源：origin: calrissian/flowmix

public void run() {
 StormTopology topology = new FlowmixBuilder()
   .setFlowLoader(new SimpleFlowLoaderSpout(provider.getFlows(), 60000))
   .setEventsLoader(new MockEventGeneratorSpout(getMockEvents(), 10))
   .setOutputBolt(new PrinterBolt())
   .setParallelismHint(6)
  .create()
 .createTopology();
 Config conf = new Config();
 conf.setNumWorkers(20);
 conf.setMaxSpoutPending(5000);
 conf.setDebug(false);
 conf.registerSerialization(BaseEvent.class, EventSerializer.class);
 conf.setSkipMissingKryoRegistrations(false);
 LocalCluster cluster = new LocalCluster();
 cluster.submitTopology("example-topology", conf, topology);
}

内容来源于网络，如有侵权，请联系作者删除！

Config

相关文章

    查看更多
    微信公众号

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新文章

    更多
    • 蜀ICP备13028337号-1 大数据知识库 https://www.saoniuhuo.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com