public static void main()

in src/main/java/com/aliyun/emr/example/spark/streaming/SparkRocketMQJavaDemo.java [42:95]
48 lines of code
2 McCabe index (conditional complexity)

    public static void main(String[] args) throws InterruptedException {
        if (args.length < 5) {
            System.err.println("Usage: spark-submit --class com.aliyun.emr.example.spark.streaming.SparkRocketMQJavaDemo " +
                "examples-1.0-SNAPSHOT-shaded.jar <accessKeyId> <accessKeySecret> " +
                "<consumerId> <topic> <subExpression>");
            System.exit(1);
        }

        String accessKeyId = args[0];
        String accessKeySecret = args[1];
        String consumerId = args[2];
        String topic = args[3];
        String subExpression = args[4];

        SparkConf sparkConf = new SparkConf().setAppName("E-MapReduce Demo 4-2: Spark RocketMQ Demo (Java)");
        // Create the context with 2 seconds batch size
        JavaStreamingContext jssc = new JavaStreamingContext(sparkConf, new Duration(2000));

        JavaReceiverInputDStream<byte[]> lines = OnsUtils.createStream(jssc, consumerId, topic, subExpression,
                accessKeyId, accessKeySecret, StorageLevel.MEMORY_AND_DISK(), new Function<Message, byte[]>() {
                    @Override
                    public byte[] call(Message msg) throws Exception {
                        return msg.getBody();
                    }
                });

        JavaDStream<String> words = lines.map(new Function<byte[], String>() {
            @Override
            public String call(byte[] v1) throws Exception {
                return new String(v1);
            }
        }).flatMap(new FlatMapFunction<String, String>() {
            @Override
            public Iterator<String> call(String x) {
                return Lists.newArrayList(SPACE.split(x)).iterator();
            }
        });
        JavaPairDStream<String, Integer> wordCounts = words.mapToPair(
                new PairFunction<String, String, Integer>() {
                    @Override
                    public Tuple2<String, Integer> call(String s) {
                        return new Tuple2<String, Integer>(s, 1);
                    }
                }).reduceByKey(new Function2<Integer, Integer, Integer>() {
            @Override
            public Integer call(Integer i1, Integer i2) {
                return i1 + i2;
            }
        });

        wordCounts.print();
        jssc.start();
        jssc.awaitTermination();
    }