Spark BIGDATA Processing Sample

-


Java based code on calcuating the word cound - 

 

 import scala.Tuple2;

public class WordCountingApp {

public static void main(String[] args) throws InterruptedException {

        Logger.getLogger("org")

            .setLevel(Level.OFF);

        Logger.getLogger("akka")

            .setLevel(Level.OFF);

 

        Map<String, Object> kafkaParams = new HashMap<>();

        kafkaParams.put("bootstrap.servers", "localhost:9092");

        kafkaParams.put("key.deserializer", StringDeserializer.class);

        kafkaParams.put("value.deserializer", StringDeserializer.class);

        kafkaParams.put("group.id", "use_a_separate_group_id_for_each_stream");

        kafkaParams.put("auto.offset.reset", "latest");

        kafkaParams.put("enable.auto.commit", false);


        Collection<String> topics = Arrays.asList("messages");


        SparkConf sparkConf = new SparkConf();

        sparkConf.setMaster("local[2]");

        sparkConf.setAppName("WordCountingApp");

        sparkConf.set("spark.cassandra.connection.host", "127.0.0.1");


        JavaStreamingContext streamingContext = new JavaStreamingContext(sparkConf, Durations.seconds(1));


        JavaInputDStream<ConsumerRecord<String, String>> messages = KafkaUtils.createDirectStream(streamingContext, LocationStrategies.PreferConsistent(), ConsumerStrategies.<String, String> Subscribe(topics, kafkaParams));


        JavaPairDStream<String, String> results = messages.mapToPair(record -> new Tuple2<>(record.key(), record.value()));


        JavaDStream<String> lines = results.map(tuple2 -> tuple2._2());


        JavaDStream<String> words = lines.flatMap(x -> Arrays.asList(x.split("\\s+"))

            .iterator());


        JavaPairDStream<String, Integer> wordCounts = words.mapToPair(s -> new Tuple2<>(s, 1))

            .reduceByKey((i1, i2) -> i1 + i2);


        wordCounts.foreachRDD(javaRdd -> {

            Map<String, Integer> wordCountMap = javaRdd.collectAsMap();

            for (String key : wordCountMap.keySet()) {

                List<Word> wordList = Arrays.asList(new Word(key, wordCountMap.get(key)));

                JavaRDD<Word> rdd = streamingContext.sparkContext()

                    .parallelize(wordList);

                javaFunctions(rdd).writerBuilder("vocabulary", "words", mapToRow(Word.class))

                    .saveToCassandra();

            }

        });


        streamingContext.start();

        streamingContext.awaitTermination();

    }

}


Comments

Post a Comment

Popular posts from this blog

Spark Cluster

-
Image
Spark applications run as independent sets of processes on a cluster, coordinated by the   SparkContext   object in your main program (called the   driver program ). Specifically, to run on a cluster, the SparkContext can connect to several types of  cluster managers  (either Spark’s own standalone cluster manager, Mesos or YARN), which allocate resources across applications. Once connected, Spark acquires  executors  on nodes in the cluster, which are processes that run computations and store data for your application. Next, it sends your application code (defined by JAR or Python files passed to SparkContext) to the executors. Finally, SparkContext sends  tasks  to the executors to run. There are several useful things to note about this architecture: Each application gets its own executor processes, which stay up for the duration of the whole application and run tasks in multiple threads. This has the benefit of isolating applications from ...
Read more

DORA Metrics

-
 DORA metrics, often referred to as the DORA (DevOps Research and Assessment) metrics, are a set of key performance indicators (KPIs) used to measure and assess the effectiveness of an organization's DevOps practices. DORA metrics were developed by the DevOps Research and Assessment team, which includes notable figures in the DevOps community such as Dr. Nicole Forsgren and Jez Humble. The DORA metrics encompass four key areas: 1. **Deployment Frequency:** This metric measures how often an organization deploys code changes to production. Higher deployment frequencies indicate a more agile and responsive development process. 2. **Lead Time for Changes:** This metric tracks the time it takes for code changes to go from code commit to being deployed in production. Shorter lead times suggest efficient and streamlined development and delivery processes. 3. **Mean Time to Recover (MTTR):** MTTR measures how quickly an organization can recover from failures or incidents in production. Low...
Read more