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

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

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

Config.get介绍

暂无

代码示例

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

@Override
public int run(String[] args) throws Exception {
  Config conf = getConf();
  if (conf.get(Config.STORM_CLUSTER_MODE).equals("rpc")) {
    SetDPRCTopology();
    return 0;
  } else if (conf.get(Config.STORM_CLUSTER_MODE).equals("local")) {
    SetLocalTopology();
    return 0;
  } else {
    SetRemoteTopology();
    return 0;
  }
}

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

private List getRegisteredSerializations() {
    if(!containsKey(Config.TOPOLOGY_KRYO_REGISTER)) {
      put(Config.TOPOLOGY_KRYO_REGISTER, new ArrayList());
    }
    return (List) get(Config.TOPOLOGY_KRYO_REGISTER);
  }
}

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

public static void runTopologyRemotely(StormTopology topology, String topologyName, Config conf,
                    int runtimeInSeconds, Callback callback) throws Exception {
  if (conf.get(Config.TOPOLOGY_WORKERS) == null) {
    conf.setNumWorkers(3);
  }
  StormSubmitter.submitTopology(topologyName, conf, topology);
  if (JStormUtils.parseBoolean(conf.get("RUN_LONG_TIME"), false)) {
    LOG.info(topologyName + " will run long time");
    return;
  }
  if (runtimeInSeconds < 120) {
    JStormUtils.sleepMs(120 * 1000);
  } else {
    JStormUtils.sleepMs(runtimeInSeconds * 1000);
  }
  if (callback != null) {
    callback.execute(topologyName);
  }
  killTopology(conf, topologyName);
}

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

public static void test() {
  TopologyBuilder builder = new TopologyBuilder();
  
  int spoutNum = JStormUtils.parseInt(conf.get("spout.num"), 8);
  int countNum = JStormUtils.parseInt(conf.get("count.num"), 8);
  builder.setSpout("spout", new InOrderSpout(), spoutNum);
  builder.setBolt("count", new Check(), countNum).fieldsGrouping("spout", new Fields("c1"));
  
  String[] className = Thread.currentThread().getStackTrace()[1].getClassName().split("\\.");
  String topologyName = className[className.length - 1];
  try {
    JStormHelper.runTopology(builder.createTopology(), 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 TopologyBuilder setBuilder() {
  BatchTopologyBuilder topologyBuilder = new BatchTopologyBuilder(topologyName);
  int spoutParallel = JStormUtils.parseInt(conf.get("topology.spout.parallel"), 1);
  BoltDeclarer boltDeclarer = topologyBuilder.setSpout("Spout", new SimpleSpout(), spoutParallel);
  int boltParallel = JStormUtils.parseInt(conf.get("topology.bolt.parallel"), 2);
  topologyBuilder.setBolt("Bolt", new SimpleBolt(), boltParallel).shuffleGrouping("Spout");
  return topologyBuilder.getTopologyBuilder();
}

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

public static void main(String[] args) throws AlreadyAliveException, InvalidTopologyException {
    Config conf = JStormHelper.getConfig(args);
    int spoutParallelism = JStormUtils.parseInt(conf.get(SPOUT_PARALLELISM_HINT), 1);
    int splitParallelism = JStormUtils.parseInt(conf.get(SPLIT_PARALLELISM_HINT), 2);
    int countParallelism = JStormUtils.parseInt(conf.get(COUNT_PARALLELISM_HINT), 2);
    boolean isValueSpout = JStormUtils.parseBoolean(conf.get("is.value.spout"), false);

    TransactionTopologyBuilder builder = new TransactionTopologyBuilder();
    if (isValueSpout)
      builder.setSpoutWithAck("spout", new BatchAckerValueSpout(), spoutParallelism);
    else
      builder.setSpoutWithAck("spout", new BatchAckerSpout(), spoutParallelism);
    builder.setBoltWithAck("split", new BatchAckerSplit(), splitParallelism).localOrShuffleGrouping("spout");;
    builder.setBoltWithAck("count", new BatchAckerCount(), countParallelism).fieldsGrouping("split", new Fields("word"));
    
    String[] className = Thread.currentThread().getStackTrace()[1].getClassName().split("\\.");
    String topologyName = className[className.length - 1];
    StormSubmitter.submitTopology(topologyName, conf, builder.createTopology());
  }
}

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

public static void test() {
  int spout_Parallelism_hint = JStormUtils.parseInt(conf.get(TOPOLOGY_SPOUT_PARALLELISM_HINT), 1);
  int split_Parallelism_hint = JStormUtils.parseInt(conf.get(TOPOLOGY_SPLIT_PARALLELISM_HINT), 1);
  int count_Parallelism_hint = JStormUtils.parseInt(conf.get(TOPOLOGY_COUNT_PARALLELISM_HINT), 2);
  
  TopologyBuilder builder = new TopologyBuilder();        
  builder.setSpout("spout", new FastRandomSentenceSpout(), spout_Parallelism_hint);
  builder.setBolt("split", new SplitSentence(), split_Parallelism_hint).localOrShuffleGrouping("spout");
  builder.setBolt("count", new WordCount(), count_Parallelism_hint).fieldsGrouping("split", new Fields("word"));
  
  String[] className = Thread.currentThread().getStackTrace()[1].getClassName().split("\\.");
  String topologyName = className[className.length - 1];
  
  isLocal = JStormHelper.localMode(conf);
  try {
    JStormHelper.runTopology(builder.createTopology(), topologyName, conf, 60,
        new JStormHelper.CheckAckedFail(conf), isLocal);
  } catch (Exception e) {
    e.printStackTrace();
    Assert.fail("Failed");
  }
}

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

public void SetRemoteTopology() throws AlreadyAliveException, InvalidTopologyException, TopologyAssignException {
  Config conf = getConf();
  StormTopology topology = buildTopology();
  
  conf.put(Config.STORM_CLUSTER_MODE, "distributed");
  String streamName = (String) conf.get(Config.TOPOLOGY_NAME);
  if (streamName == null) {
    streamName = "SequenceTest";
  }
  
  if (streamName.contains("zeromq")) {
    conf.put(Config.STORM_MESSAGING_TRANSPORT, "com.alibaba.jstorm.message.zeroMq.MQContext");
    
  } else {
    conf.put(Config.STORM_MESSAGING_TRANSPORT, "com.alibaba.jstorm.message.netty.NettyContext");
  }
  
  StormSubmitter.submitTopology(streamName, conf, topology);
  
}

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

public static void test() {
  int spout_Parallelism_hint = JStormUtils.parseInt(conf.get(TOPOLOGY_SPOUT_PARALLELISM_HINT), 1);
  int count_Parallelism_hint = JStormUtils.parseInt(conf.get(TOPOLOGY_COUNT_PARALLELISM_HINT), 1);
  TopologyBuilder builder = new TopologyBuilder();
  builder.setSpout("spout", new FastRandomSentenceSpout(), spout_Parallelism_hint);
  WordCount wordCountBolt = new WordCount();
  builder.setBolt("count", wordCountBolt.sessionTimeWindow(Time.seconds(1L))
      .withWindowStateMerger(wordCountBolt), count_Parallelism_hint)
      .fieldsGrouping("spout", new Fields("word"));
  //.allGrouping("spout", Common.WATERMARK_STREAM_ID);
  String[] className = Thread.currentThread().getStackTrace()[1].getClassName().split("\\.");
  String topologyName = className[className.length - 1];
  try {
    JStormHelper.runTopology(builder.createTopology(), topologyName, conf, 60,
        new JStormHelper.CheckAckedFail(conf), true);
  } catch (Exception e) {
    e.printStackTrace();
  }
}

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

public static void test() throws Exception {
  int spout_Parallelism_hint = JStormUtils.parseInt(conf.get(TOPOLOGY_SPOUT_PARALLELISM_HINT), 1);
  int count_Parallelism_hint = JStormUtils.parseInt(conf.get(TOPOLOGY_COUNT_PARALLELISM_HINT), 1);
  TopologyBuilder builder = new TopologyBuilder();
  builder.setSpout("spout", new FastRandomSentenceSpout(), spout_Parallelism_hint);
  WordCount wordCountBolt = new WordCount();
  builder.setBolt("count", wordCountBolt.sessionEventTimeWindow(Time.milliseconds(3L))
      .withTimestampExtractor(wordCountBolt)
      .withWindowStateMerger(wordCountBolt), count_Parallelism_hint)
      .fieldsGrouping("spout", new Fields("word", "ts"));
  String[] className = Thread.currentThread().getStackTrace()[1].getClassName().split("\\.");
  String topologyName = className[className.length - 1];
  JStormHelper.runTopology(builder.createTopology(), topologyName, conf, 60,
      new JStormHelper.CheckAckedFail(conf), true);
}

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

public static void test(Config conf) throws Exception {
  int spout_Parallelism_hint = JStormUtils.parseInt(conf.get(TOPOLOGY_SPOUT_PARALLELISM_HINT), 1);
  int split_Parallelism_hint = JStormUtils.parseInt(conf.get(TOPOLOGY_SPLIT_PARALLELISM_HINT), 1);
  int count_Parallelism_hint = JStormUtils.parseInt(conf.get(TOPOLOGY_COUNT_PARALLELISM_HINT), 1);
  TopologyBuilder builder = new TopologyBuilder();
  builder.setSpout("spout", new FastRandomSentenceSpout(), spout_Parallelism_hint);
  builder.setBolt("split", new SplitSentence(), split_Parallelism_hint).shuffleGrouping("spout");
  long windowSize = JStormUtils.parseLong(conf.get("window.size.sec"), 10);
  long stateWindowSize = JStormUtils.parseLong(conf.get("state.window.size.sec"), 60);
  builder.setBolt("count", new WordCount()
          .timeWindow(Time.seconds(windowSize))
          .withStateSize(Time.seconds(stateWindowSize)),
      count_Parallelism_hint).fieldsGrouping("split", new Fields("word"));
  String[] className = Thread.currentThread().getStackTrace()[1].getClassName().split("\\.");
  String topologyName = className[className.length - 1];
  boolean isLocal = true;
  if (conf.containsKey("storm.cluster.mode")) {
    isLocal = StormConfig.local_mode(conf);
  }
  // RUN_LONG_TIME = true
  JStormHelper.runTopology(builder.createTopology(), topologyName, conf, 60,
      new JStormHelper.CheckAckedFail(conf), isLocal);
}

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

public static void test() {
  int spout_Parallelism_hint = JStormUtils.parseInt(conf.get(TOPOLOGY_SPOUT_PARALLELISM_HINT), 1);
  int split_Parallelism_hint = JStormUtils.parseInt(conf.get(TOPOLOGY_SPLIT_PARALLELISM_HINT), 1);
  int count_Parallelism_hint = JStormUtils.parseInt(conf.get(TOPOLOGY_COUNT_PARALLELISM_HINT), 1);
  boolean isLocalShuffle = JStormUtils.parseBoolean(conf.get("is.local.first.group"), false);

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

public static void test() throws Exception {
  TransactionTopologyBuilder builder = new TransactionTopologyBuilder();
  if (isLocal) {
    conf.put("tuple.num.per.batch", 5);
    conf.put("transaction.scheduler.spout", false);
    conf.put("transaction.exactly.cache.type", "default");
  }
  int spoutParallelism = JStormUtils.parseInt(conf.get(SPOUT_PARALLELISM_HINT), 1);
  int splitParallelism = JStormUtils.parseInt(conf.get(SPLIT_PARALLELISM_HINT), 2);
  int countParallelism = JStormUtils.parseInt(conf.get(COUNT_PARALLELISM_HINT), 2);
  boolean isScheduleSpout = JStormUtils.parseBoolean(conf.get("transaction.scheduler.spout"), true);
  if (isScheduleSpout)
    // Generate batch by configured time. "transaction.schedule.batch.delay.ms: 1000 # 1sec"
    builder.setSpout("spout", new ScheduleTxSpout(), spoutParallelism);
  else
    // Generate batch by user when calling emitBarrier
    builder.setSpout("spout", new BasicTxSpout(), spoutParallelism, false);
  builder.setBolt("split", new TxSplitSentence(), splitParallelism).localOrShuffleGrouping("spout");
  builder.setBolt("count", new TxWordCount(), countParallelism).fieldsGrouping("split", new Fields("word"));
  String[] className = Thread.currentThread().getStackTrace()[1].getClassName().split("\\.");
  String topologyName = className[className.length - 1];
  StormSubmitter.submitTopology(topologyName, conf, builder.createTopology());
}

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

public static void test() throws Exception {
  int spout_parallelism_hint = JStormUtils.parseInt(conf.get(TOPOLOGY_SPOUT_PARALLELISM_HINT), 1);
  int split_parallelism_hint = JStormUtils.parseInt(conf.get(TOPOLOGY_SPLIT_PARALLELISM_HINT), 1);
  int count_parallelism_hint = JStormUtils.parseInt(conf.get(TOPOLOGY_COUNT_PARALLELISM_HINT), 1);
  TopologyBuilder builder = new TopologyBuilder();
  builder.setSpout("spout", new FastRandomSentenceSpout(), spout_parallelism_hint);
  builder.setBolt("split", new SplitSentence(), split_parallelism_hint).shuffleGrouping("spout");
  builder.setBolt("count", new WordCount()
          .ingestionTimeWindow(Time.seconds(1L), Time.milliseconds(500L))
          .withStateSize(Time.minutes(10)),
      count_parallelism_hint).fieldsGrouping("split", new Fields("word"));
  String[] className = Thread.currentThread().getStackTrace()[1].getClassName().split("\\.");
  String topologyName = className[className.length - 1];
  JStormHelper.runTopology(builder.createTopology(), topologyName, conf, 60,
      new JStormHelper.CheckAckedFail(conf), true);
}

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

int spout_Parallelism_hint = JStormUtils.parseInt(conf.get(TOPOLOGY_SPOUT_PARALLELISM_HINT), 1);
int split_Parallelism_hint = JStormUtils.parseInt(conf.get(TOPOLOGY_SPLIT_PARALLELISM_HINT), 1);
int count_Parallelism_hint = JStormUtils.parseInt(conf.get(TOPOLOGY_COUNT_PARALLELISM_HINT), 2);

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

public static void test() {
  int spout_Parallelism_hint = JStormUtils.parseInt(conf.get(TOPOLOGY_SPOUT_PARALLELISM_HINT), 1);
  int split_Parallelism_hint = JStormUtils.parseInt(conf.get(TOPOLOGY_SPLIT_PARALLELISM_HINT), 1);
  int count_Parallelism_hint = JStormUtils.parseInt(conf.get(TOPOLOGY_COUNT_PARALLELISM_HINT), 1);
  TopologyBuilder builder = new TopologyBuilder();
  boolean isLocalShuffle = JStormUtils.parseBoolean(conf.get("is.local.first.group"), false);
  builder.setSpout("spout", new FastRandomSentenceSpout(), spout_Parallelism_hint);
  if (isLocalShuffle) {
    builder.setBolt("split", new SplitSentence(), split_Parallelism_hint).localFirstGrouping("spout");
  } else {
    builder.setBolt("split", new SplitSentence(), split_Parallelism_hint).shuffleGrouping("spout");
  }
  builder.setBolt("count", new WordCount()
          .timeWindow(Time.seconds(1L), Time.milliseconds(500L))
          .withStateSize(Time.hours(2)),
      count_Parallelism_hint).fieldsGrouping("split", new Fields("word"));
  String[] className = Thread.currentThread().getStackTrace()[1].getClassName().split("\\.");
  String topologyName = className[className.length - 1];
  try {
    JStormHelper.runTopology(builder.createTopology(), topologyName, conf, 60,
        new JStormHelper.CheckAckedFail(conf), true);
  } catch (Exception e) {
    e.printStackTrace();
  }
}

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

public static void test() throws Exception {
  int spout_Parallelism_hint = JStormUtils.parseInt(conf.get(TOPOLOGY_SPOUT_PARALLELISM_HINT), 1);
  int count_Parallelism_hint = JStormUtils.parseInt(conf.get(TOPOLOGY_COUNT_PARALLELISM_HINT), 1);
  TopologyBuilder builder = new TopologyBuilder();
  builder.setSpout("spout", new FastRandomSentenceSpout(), spout_Parallelism_hint);
  WordCount wordCountBolt = new WordCount();
  builder.setBolt("count", wordCountBolt.eventTimeWindow(Time.milliseconds(3L))
      .withTimestampExtractor(wordCountBolt)
      .withWatermarkGenerator(new PeriodicWatermarkGenerator(Time.milliseconds(1L), Time.milliseconds(10L)))
      , count_Parallelism_hint)
      .fieldsGrouping("spout", new Fields("word", "ts"));
  String[] className = Thread.currentThread().getStackTrace()[1].getClassName().split("\\.");
  String topologyName = className[className.length - 1];
  JStormHelper.runTopology(builder.createTopology(), topologyName, conf, 60,
      new JStormHelper.CheckAckedFail(conf), true);
}

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

public static void test() throws Exception {
  TransactionTopologyBuilder builder = new TransactionTopologyBuilder();
  if (isLocal) {
    conf.put("tuple.num.per.batch", 100);
    conf.put("transaction.scheduler.spout", false);
    conf.put("transaction.exactly.cache.type", "default");
    conf.put("transaction.topology", true);
  }
  int spoutParallelismHint = JStormUtils.parseInt(conf.get(TOPOLOGY_SPOUT_PARALLELISM_HINT), 1);
  int splitParallelismHint = JStormUtils.parseInt(conf.get(TOPOLOGY_SPLIT_PARALLELISM_HINT), 1);
  int countParallelismHint = JStormUtils.parseInt(conf.get(TOPOLOGY_COUNT_PARALLELISM_HINT), 1);
  builder.setSpout("spout", new TxFastRandomSentenceSpout(), spoutParallelismHint);
  builder.setBolt("split", new TxSplitSentence(), splitParallelismHint).localOrShuffleGrouping("spout");
  WordCount wordCount = new WordCount();
  builder.setBolt("count", wordCount
          .timeWindow(Time.seconds(60L))
          .withTransactionStateOperator(wordCount),
      countParallelismHint).fieldsGrouping("split", new Fields("word"));
  builder.enableHdfs();
  String[] className = Thread.currentThread().getStackTrace()[1].getClassName().split("\\.");
  String topologyName = className[className.length - 1];
  StormSubmitter.submitTopology(topologyName, conf, builder.createTopology());
}

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

public static void test() {
  int spout_Parallelism_hint = JStormUtils.parseInt(conf.get(TOPOLOGY_SPOUT_PARALLELISM_HINT), 1);
  int split_Parallelism_hint = JStormUtils.parseInt(conf.get(TOPOLOGY_SPLIT_PARALLELISM_HINT), 1);
  int count_Parallelism_hint = JStormUtils.parseInt(conf.get(TOPOLOGY_COUNT_PARALLELISM_HINT), 1);
  TopologyBuilder builder = new TopologyBuilder();
  builder.setSpout("spout", new FastRandomSentenceSpout(), spout_Parallelism_hint);
  builder.setBolt("split", new SplitSentence(), split_Parallelism_hint).shuffleGrouping("spout");
  int topN = 10;
  Time win = Time.seconds(10L);
  builder.setBolt("count", new WordCount(topN)
          .timeWindow(win)
          .withStateSize(Time.seconds(120L)),
      count_Parallelism_hint).fieldsGrouping("split", new Fields("word"));
  builder.setBolt("merge",
      new MergeTopN(topN).timeWindow(win), 1).allGrouping("count");
  String[] className = Thread.currentThread().getStackTrace()[1].getClassName().split("\\.");
  String topologyName = className[className.length - 1];
  try {
    JStormHelper.runTopology(builder.createTopology(), topologyName, conf, 60,
        new JStormHelper.CheckAckedFail(conf), true);
  } catch (Exception e) {
    e.printStackTrace();
  }
}

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

public static void test() throws Exception {
  TransactionTopologyBuilder builder = new TransactionTopologyBuilder();
  if (isLocal) {
    conf.put("tuple.num.per.batch", 100);
    conf.put("transaction.scheduler.spout", false);
    conf.put("transaction.exactly.cache.type", "default");
    conf.put("transaction.topology", true);
  }
  int spoutParallelismHint = JStormUtils.parseInt(conf.get(TOPOLOGY_SPOUT_PARALLELISM_HINT), 1);
  int splitParallelismHint = JStormUtils.parseInt(conf.get(TOPOLOGY_SPLIT_PARALLELISM_HINT), 1);
  int countParallelismHint = JStormUtils.parseInt(conf.get(TOPOLOGY_COUNT_PARALLELISM_HINT), 1);
  builder.setSpout("spout", new TxFastRandomSentenceSpout(), spoutParallelismHint);
  builder.setBolt("split", new SplitSentence(), splitParallelismHint).localOrShuffleGrouping("spout");
  builder.setBolt("count", new WordCount()
          .timeWindow(Time.seconds(1L))
          .withStateSize(Time.hours(2)),
      countParallelismHint).fieldsGrouping("split", new Fields("word"));
  String[] className = Thread.currentThread().getStackTrace()[1].getClassName().split("\\.");
  String topologyName = className[className.length - 1];
  JStormHelper.runTopology(builder.createTopology(), topologyName, conf, 60,
      new JStormHelper.CheckAckedFail(conf), true);
}

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

Config

相关文章

    查看更多

    热门标签

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

    最新文章

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