import org.apache.cassandra.hadoop.cql3.CqlOutputFormat; //导入依赖的package包/类
public int run(String[] args) throws Exception
{
    String outputReducerType = "filesystem";
    if (args != null && args[0].startsWith(OUTPUT_REDUCER_VAR))
    {
        String[] s = args[0].split("=");
        if (s != null && s.length == 2)
            outputReducerType = s[1];
    }
    logger.info("output reducer type: " + outputReducerType);

    Job job = new Job(getConf(), "wordcount");
    job.setJarByClass(WordCount.class);
    job.setMapperClass(TokenizerMapper.class);

    if (outputReducerType.equalsIgnoreCase("filesystem"))
    {
        job.setCombinerClass(ReducerToFilesystem.class);
        job.setReducerClass(ReducerToFilesystem.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        FileOutputFormat.setOutputPath(job, new Path(OUTPUT_PATH_PREFIX));
    }
    else
    {
        job.setReducerClass(ReducerToCassandra.class);

        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);
        job.setOutputKeyClass(Map.class);
        job.setOutputValueClass(List.class);

        job.setOutputFormatClass(CqlOutputFormat.class);

        ConfigHelper.setOutputColumnFamily(job.getConfiguration(), KEYSPACE, OUTPUT_COLUMN_FAMILY);
        job.getConfiguration().set(PRIMARY_KEY, "word,sum");
        String query = "UPDATE " + KEYSPACE + "." + OUTPUT_COLUMN_FAMILY +
                       " SET count_num = ? ";
        CqlConfigHelper.setOutputCql(job.getConfiguration(), query);
        ConfigHelper.setOutputInitialAddress(job.getConfiguration(), "localhost");
        ConfigHelper.setOutputPartitioner(job.getConfiguration(), "Murmur3Partitioner");
    }

    job.setInputFormatClass(CqlPagingInputFormat.class);

    ConfigHelper.setInputRpcPort(job.getConfiguration(), "9160");
    ConfigHelper.setInputInitialAddress(job.getConfiguration(), "localhost");
    ConfigHelper.setInputColumnFamily(job.getConfiguration(), KEYSPACE, COLUMN_FAMILY);
    ConfigHelper.setInputPartitioner(job.getConfiguration(), "Murmur3Partitioner");

    CqlConfigHelper.setInputCQLPageRowSize(job.getConfiguration(), "3");
    //this is the user defined filter clauses, you can comment it out if you want count all titles
    CqlConfigHelper.setInputWhereClauses(job.getConfiguration(), "title='A'");
    job.waitForCompletion(true);
    return 0;
}