code: public static void main(String[] args) throws Exception {
      if (args == null || args.length == 0 || args[0] == null || args[0].trim().length() == 0) {
        System.err.println("Invalid command-line args! Must pass the name of a processor to run.\n"
            + "Supported processors:\n");
        displayProcessorOptions(System.err);
        System.exit(1);
      }
  
      // Determine the processor to run
      RDDProcessor procImpl;
      ClassLoader myCL = SparkApp.class.getClassLoader();
      try {
        Class<? extends RDDProcessor> clazz = (Class<? extends RDDProcessor>) myCL.loadClass(args[0]);
        procImpl = clazz.newInstance();
      } catch (ClassNotFoundException cnfe) {
        procImpl = newProcessor(args[0].trim().toLowerCase(Locale.ROOT));
      }
  
      // ensure the processor is serializable
      assertSerializable(procImpl);
  
      String[] procImplArgs = new String[args.length - 1];
      System.arraycopy(args, 1, procImplArgs, 0, procImplArgs.length);
  
      // process command-line args to configure this application
      CommandLine cli =
          processCommandLineArgs(
            joinCommonAndProcessorOptions(procImpl.getOptions()), procImplArgs);
  
      SparkConf sparkConf = new SparkConf().setAppName(procImpl.getName());
  
      //sparkConf.set("spark.serializer", KryoSerializer.class.getName());
      //sparkConf.set("spark.kryo.registrator", LWKryoRegistrator.class.getName());
      sparkConf.set("spark.task.maxFailures", "10");
  
      setupSolrAuthenticationProps(cli, sparkConf);
  
      String masterUrl = cli.getOptionValue("master");
      if (masterUrl != null)
        sparkConf.setMaster(masterUrl);
  
      // Create a local StreamingContext with two working thread and batch interval
      log.info("Running processor "+procImpl.getName());
      int exitCode = procImpl.run(sparkConf, cli);
  
      System.exit(exitCode);
    }
docs: Runs a stream processor implementation.