import storm.trident.state.StateUpdater; //导入依赖的package包/类
public static void main(String[] args) throws AlreadyAliveException, InvalidTopologyException {
    if (args.length < 3) {
        Logger.getAnonymousLogger().log(Level.SEVERE, "where are the commandline args? -- use args -- folder numWorkers windowSize");
        System.exit(-1);
    }

    final FieldTemplate template = new MlStormFieldTemplate();
    final int numWorkers = Integer.valueOf(args[1]);
    final int windowSize = Integer.valueOf(args[2]);
    final StateUpdater stateUpdater = new CobwebClusterUpdater(template);
    final StateFactory stateFactory = new MlStormClustererFactory.CobwebClustererFactory(numWorkers, windowSize);
    final QueryFunction<CobwebClustererState, String> queryFunction = new MlStormClustererQuery.CobwebClustererQuery();
    final MlStormSpout features = new MddbFeatureExtractorSpout(args[0], template);
    final StormTopology stormTopology = WekaBaseLearningTopology.buildTopology(features, template, numWorkers, stateUpdater, stateFactory, queryFunction, null, TOPOLOGY_DRPC_NAME, null);

    if (numWorkers == 1) {
        LocalCluster cluster = new LocalCluster();
        cluster.submitTopology(TOPOLOGY_DRPC_NAME, MlStormConfig.getDefaultMlStormConfig(numWorkers), stormTopology);
    } else {
        StormSubmitter.submitTopology(TOPOLOGY_DRPC_NAME, MlStormConfig.getDefaultMlStormConfig(numWorkers), stormTopology);
    }
}
 

import storm.trident.state.StateUpdater; //导入依赖的package包/类
public static void main(String[] args) throws AlreadyAliveException, InvalidTopologyException {
    if (args.length < 5) {
        System.err.println(" Where are all the arguments? -- use args -- folder numWorkers windowSize k parallelism");
        return;
    }

    final FieldTemplate template = new MlStormFieldTemplate();
    final int numWorkers = Integer.valueOf(args[1]);
    final int windowSize = Integer.valueOf(args[2]);
    final int k = Integer.valueOf(args[3]);
    final int parallelism = Integer.valueOf(args[4]);
    final StateUpdater stateUpdater = new KmeansClusterUpdater(template);
    final StateFactory stateFactory = new MlStormClustererFactory.KmeansClustererFactory(k, windowSize, template);
    final QueryFunction<KmeansClustererState, String> queryFunction = new MlStormClustererQuery.KmeansClustererQuery();
    final QueryFunction<KmeansClustererState, String> parameterUpdateFunction = new MlStormClustererQuery.KmeansNumClustersUpdateQuery();
    final MlStormSpout features = new MddbFeatureExtractorSpout(args[0], template);
    final StormTopology stormTopology = buildTopology(features, template, parallelism, stateUpdater, stateFactory, queryFunction, parameterUpdateFunction, "kmeans", "kUpdate");

    if (numWorkers == 1) {
        LocalCluster cluster = new LocalCluster();
        cluster.submitTopology("kmeans", MlStormConfig.getDefaultMlStormConfig(numWorkers), stormTopology);
    } else {
        StormSubmitter.submitTopology("kmeans", MlStormConfig.getDefaultMlStormConfig(numWorkers), stormTopology);
    }
}
 

import storm.trident.state.StateUpdater; //导入依赖的package包/类
public static void main(String[] args) throws Exception {
    if (args.length < 4) {
        Logger.getAnonymousLogger().log(Level.SEVERE, " Where are all the arguments? -- use args -- file numWorkers windowSize parallelism");
        System.exit(-1);
    }

    final FieldTemplate template = new MlStormFieldTemplate();
    final int numWorkers = Integer.valueOf(args[1]);
    final int windowSize = Integer.valueOf(args[2]);
    final int parallelism = Integer.valueOf(args[3]);
    final StateUpdater stateUpdater = new BinaryClassifierStateUpdater(template);
    final StateFactory stateFactory = new BinaryClassifierFactory(WekaClassificationAlgorithms.svm.name(), windowSize, template, null /* weka.core.Utils.splitOptions("-C 1.0 -L 0.0010 -P 1.0E-12 -N 0 -V -1 -W 1 -K \"weka.classifiers.functions.supportVector.PolyKernel -C 250007 -E 1.0\"")*/);
    final QueryFunction<MlStormWekaState, Integer> queryFunction = new BinaryClassifierQuery.SvmQuery();
    final QueryFunction<KmeansClustererState, String> parameterUpdateFunction = null;
    final MlStormSpout features = new AustralianElectricityPricingSpout(args[0], template);
    final StormTopology stormTopology = WekaBaseLearningTopology.buildTopology(features, template, parallelism, stateUpdater, stateFactory, queryFunction, parameterUpdateFunction, "svm", "svmUpdate");

    if (numWorkers == 1) {
        LocalCluster cluster = new LocalCluster();
        cluster.submitTopology("SVM", MlStormConfig.getDefaultMlStormConfig(numWorkers), stormTopology);
    } else {
        StormSubmitter.submitTopology("SVM", MlStormConfig.getDefaultMlStormConfig(numWorkers), stormTopology);
    }
}
 

import storm.trident.state.StateUpdater; //导入依赖的package包/类
public TridentState partitionPersist(StateSpec stateSpec, Fields inputFields, StateUpdater updater, Fields functionFields) {
    projectionValidation(inputFields);
    String id = _topology.getUniqueStateId();
    ProcessorNode n = new ProcessorNode(_topology.getUniqueStreamId(),
                _name,
                functionFields,
                functionFields,
                new PartitionPersistProcessor(id, inputFields, updater));
    n.committer = true;
    n.stateInfo = new NodeStateInfo(id, stateSpec);
    return _topology.addSourcedStateNode(this, n);
}
 

import storm.trident.state.StateUpdater; //导入依赖的package包/类
public TridentState partitionPersist(StateSpec stateSpec, Fields inputFields, StateUpdater updater, Fields functionFields) {
    projectionValidation(inputFields);
    String id = _topology.getUniqueStateId();
    ProcessorNode n =
            new ProcessorNode(_topology.getUniqueStreamId(), _name, functionFields, functionFields, new PartitionPersistProcessor(id, inputFields, updater));
    n.committer = true;
    n.stateInfo = new NodeStateInfo(id, stateSpec);
    return _topology.addSourcedStateNode(this, n);
}
 

import storm.trident.state.StateUpdater; //导入依赖的package包/类
private static void assertArguments(IRichSpout spout, int parallelism, List<StateUpdater> stateUpdaters, List<StateFactory> stateFactories, List<QueryFunction> queryFunctions, List<String> drpcQueryFunctionNames, ReducerAggregator drpcPartitionResultAggregator, StateFactory metaStateFactory, StateUpdater metaStateUpdater, QueryFunction metaQueryFunction) {
    assert spout != null;
    assert parallelism != 0;
    assert stateFactories != null;
    assert queryFunctions != null;
    assert drpcPartitionResultAggregator != null;
    assert stateUpdaters != null;
    assert drpcQueryFunctionNames != null;
    assert metaQueryFunction != null;
    assert metaStateFactory != null;
    assert metaStateUpdater != null;
}
 

import storm.trident.state.StateUpdater; //导入依赖的package包/类
private Stream window(WindowConfig windowConfig, WindowsStoreFactory windowStoreFactory, Fields inputFields, Aggregator aggregator,
                      Fields functionFields, boolean storeTuplesInStore) {
    projectionValidation(inputFields);
    windowConfig.validate();

    Fields fields = addTriggerField(functionFields);

    // when storeTuplesInStore is false then the given windowStoreFactory is only used to store triggers and
    // that store is passed to WindowStateUpdater to remove them after committing the batch.
    Stream stream = _topology.addSourcedNode(this,
            new ProcessorNode(_topology.getUniqueStreamId(),
                    _name,
                    fields,
                    fields,
                    new WindowTridentProcessor(windowConfig, _topology.getUniqueWindowId(), windowStoreFactory,
                            inputFields, aggregator, storeTuplesInStore)));

    Stream effectiveStream = stream.project(functionFields);

    // create StateUpdater with the given windowStoreFactory to remove triggered aggregation results form store
    // when they are successfully processed.
    StateFactory stateFactory = new WindowsStateFactory();
    StateUpdater stateUpdater = new WindowsStateUpdater(windowStoreFactory);
    stream.partitionPersist(stateFactory, new Fields(WindowTridentProcessor.TRIGGER_FIELD_NAME), stateUpdater, new Fields());

    return effectiveStream;
}
 

import storm.trident.state.StateUpdater; //导入依赖的package包/类
public PartitionPersistProcessor(String stateId, Fields inputFields, StateUpdater updater) {
    _updater = updater;
    _stateId = stateId;
    _inputFields = inputFields;
}