<commit_before>package cz.zee.mrtweety.analytic.consumer;

import org.apache.spark.SparkConf;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaPairDStream;
import org.apache.spark.streaming.api.java.JavaPairReceiverInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import org.apache.spark.streaming.kafka.KafkaUtils;
import org.json.JSONObject;
import scala.Tuple2;

import java.io.Serializable;
import java.util.Arrays;
import java.util.Collections;
import java.util.regex.Pattern;

/**
 * Consumes Kafka messages.
 * @author <NAME>
 */
public class KafkaConsumer implements Serializable {

    private static final Pattern SPACE = Pattern.compile(" ");

    public static void main(String[] args) {
        new KafkaConsumer().start();
    }

    private void start() {
        SparkConf conf = new SparkConf().setAppName("MrTweety Analytic");

        // setup processing with batch interval
        JavaStreamingContext streamingContext = new JavaStreamingContext(conf, Durations.seconds(10));

        // get the Kafka stream
        JavaPairReceiverInputDStream<String, String> messages =
                // Zookeeper location and Kafka topic
                KafkaUtils.createStream(streamingContext, "localhost:2181", "mrtweety-analytic", Collections.singletonMap("tweet", 1));

        JavaDStream<String> lines = messages.map(Tuple2::_2);<commit_msg>Add local Spark master to simplify IDE development.<commit_after>package cz.zee.mrtweety.analytic.consumer;

import org.apache.spark.SparkConf;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaPairDStream;
import org.apache.spark.streaming.api.java.JavaPairReceiverInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import org.apache.spark.streaming.kafka.KafkaUtils;
import org.json.JSONObject;
import scala.Tuple2;

import java.io.Serializable;
import java.util.Arrays;
import java.util.Collections;
import java.util.regex.Pattern;

/**
 * Consumes Kafka messages.
 * @author <NAME>
 */
public class KafkaConsumer implements Serializable {

    private static final Pattern SPACE = Pattern.compile(" ");

    public static void main(String[] args) {
        new KafkaConsumer().start();
    }

    private void start() {
        SparkConf conf = new SparkConf().setAppName("MrTweety Analytic").setMaster("local[*]");

        // setup processing with batch interval
        JavaStreamingContext streamingContext = new JavaStreamingContext(conf, Durations.seconds(10));

        // get the Kafka stream
        JavaPairReceiverInputDStream<String, String> messages =
                // Zookeeper location and Kafka topic
                KafkaUtils.createStream(streamingContext, "localhost:2181", "mrtweety-analytic", Collections.singletonMap("tweet", 1));

        JavaDStream<String> lines = messages.map(Tuple2::_2);
{'fraction_non_alphanumeric': 0.10711931443638761, 'fraction_numerical': 0.0065919578114700065, 'mean_word_length': 4.1440677966101696, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4742770', 'n_tokens_mistral': 917, 'n_tokens_neox': 862, 'n_words': 184}