Skip to content

O
orders-monitoring-dashboard
Commit e1720051 authored by Ramakanth Dhane's avatar Ramakanth Dhane
Browse files
Options

Merge branch 'omd-stream-processor-feature' into 'master' 

Committed changes to configure Spark with Kafka to store the Orders in Mongo DB

See merge request !10
parents 72b92fd7 1930cedc
Hide whitespace changes
Inline Side-by-side
Showing with 246 additions and 42 deletions
omd-stream-processor/pom.xml
View file @ e1720051
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.2.6.RELEASE</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>
<groupId>com.nisum.omd</groupId>
<artifactId>omd-stream-processor</artifactId>
<version>0.0.1-SNAPSHOT</version>
<name>omd-stream-processor</name>
<description>Demo project for Spring Boot</description>
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.2.6.RELEASE</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>
<groupId>com.nisum.omd</groupId>
<artifactId>omd-stream-processor</artifactId>
<version>0.0.1-SNAPSHOT</version>
<name>omd-stream-processor</name>
<description>Demo project for Spring Boot</description>
<properties>
<java.version>1.8</java.version>
</properties>
<properties>
<java.version>1.8</java.version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.scala-lang</groupId>
<artifactId>scala-library</artifactId>
<version>2.11.8</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-core_2.11</artifactId>
<version>2.2.1</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-streaming_2.11</artifactId>
<version>2.2.1</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-streaming-kafka-0-10_2.11</artifactId>
<version>2.2.1</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-sql_2.11</artifactId>
<version>2.2.1</version>
</dependency>
<dependency>
<groupId>org.mongodb.spark</groupId>
<artifactId>mongo-spark-connector_2.11</artifactId>
<version>2.2.1</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
<exclusions>
<exclusion>
<groupId>org.junit.vintage</groupId>
<artifactId>junit-vintage-engine</artifactId>
</exclusion>
</exclusions>
</dependency>
</dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
<exclusions>
<exclusion>
<groupId>org.junit.vintage</groupId>
<artifactId>junit-vintage-engine</artifactId>
</exclusion>
</exclusions>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
</plugin>
</plugins>
</build>
<build>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
</plugin>
</plugins>
</build>
</project>
omd-stream-processor/src/main/java/com/nisum/omd/kafka/connection/OmdKafkaConnection.java 0 → 100755
View file @ e1720051
package com.nisum.omd.kafka.connection;
import com.nisum.omd.kafka.utility.OmdKafkaUtility;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import java.util.HashMap;
import java.util.Map;
public class OmdKafkaConnection {
public static Map<String ,Object> kakfkaconfig()
{
Map<String, Object> kafkaProperties = new HashMap<>();
kafkaProperties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, OmdKafkaUtility.KAFKA_BROKERS);
kafkaProperties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
kafkaProperties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
kafkaProperties.put(ConsumerConfig.GROUP_ID_CONFIG, OmdKafkaUtility.KAFKA_GROUP);
kafkaProperties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, OmdKafkaUtility.KAFKA_OFFSET_RESET_TYPE);
kafkaProperties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
return kafkaProperties;
}
}
omd-stream-processor/src/main/java/com/nisum/omd/kafka/utility/OmdKafkaUtility.java 0 → 100755
View file @ e1720051
package com.nisum.omd.kafka.utility;
public class OmdKafkaUtility {
public static final String APPLICATION_NAME = "Streaming Order DStream";
public static final String HADOOP_HOME_DIR_VALUE = "C:/winutils";
public static final String RUN_LOCAL_WITH_AVAILABLE_CORES = "local[*]";
public static final int BATCH_DURATION_INTERVAL_MS = 50000;
public static final String KAFKA_BROKERS = "localhost:9092";
public static final String KAFKA_OFFSET_RESET_TYPE = "latest";
public static final String KAFKA_GROUP = "omdgroup";
public static final String KAFKA_TOPIC = "TOPIC_OMD_ORDER_DATA";
public static final String MONGODB_OUTPUT_URI = "mongodb://localhost/Orders.orders";
private OmdKafkaUtility() {
}
}
omd-stream-processor/src/main/java/com/nisum/omd/mongo/repository/OmdSparkStreamRepository.java 0 → 100755
View file @ e1720051
package com.nisum.omd.mongo.repository;
public interface OmdSparkStreamRepository {
void saveMongoData();
}
omd-stream-processor/src/main/java/com/nisum/omd/mongo/repository/OmdSparkStreamRepositoryImpl.java 0 → 100755
View file @ e1720051
package com.nisum.omd.mongo.repository;
import com.mongodb.spark.MongoSpark;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaInputDStream;
import org.apache.spark.streaming.kafka010.CanCommitOffsets;
import org.apache.spark.streaming.kafka010.HasOffsetRanges;
import org.apache.spark.streaming.kafka010.OffsetRange;
import org.bson.Document;
public class OmdSparkStreamRepositoryImpl {
public static void saveMongoData(JavaInputDStream<ConsumerRecord<String, String>> orderStream) {
JavaDStream<ConsumerRecord<String, String>> recordJavaDStream =
orderStream.filter(f -> !f.value().contains("\"guests\":0"));
recordJavaDStream.foreachRDD((JavaRDD<ConsumerRecord<String, String>> r) -> {
MongoSpark.save(
r.map(
e -> Document.parse(e.value())
)
);
});
orderStream.foreachRDD((JavaRDD<ConsumerRecord<String, String>> javaRDD) -> {
OffsetRange[] offsetRanges = ((HasOffsetRanges) javaRDD.rdd()).offsetRanges();
((CanCommitOffsets) orderStream.inputDStream())
.commitAsync(offsetRanges, new OrderOffsetCommitCallback());
});
}
}
omd-stream-processor/src/main/java/com/nisum/omd/mongo/repository/OrderOffsetCommitCallback.java 0 → 100755
View file @ e1720051
package com.nisum.omd.mongo.repository;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.clients.consumer.OffsetCommitCallback;
import org.apache.kafka.common.TopicPartition;
import java.util.Map;
import java.util.logging.Level;
import java.util.logging.Logger;
public class OrderOffsetCommitCallback implements OffsetCommitCallback {
private static final Logger log = Logger.getLogger(OrderOffsetCommitCallback.class.getName());
@Override
public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception exception) {
log.info("---------------------------------------------------");
log.log(Level.INFO, "{0} | {1}", new Object[]{offsets, exception});
log.info("---------------------------------------------------");
}
}
omd-stream-processor/src/main/java/com/nisum/omd/spark/service/OmdSparkStreamService.java 0 → 100755
View file @ e1720051
package com.nisum.omd.spark.service;
import com.nisum.omd.kafka.utility.OmdKafkaUtility;
import com.nisum.omd.kafka.connection.OmdKafkaConnection;
import com.nisum.omd.mongo.repository.OmdSparkStreamRepositoryImpl;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.spark.SparkConf;
import org.apache.spark.streaming.Duration;
import org.apache.spark.streaming.api.java.JavaInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import org.apache.spark.streaming.kafka010.ConsumerStrategies;
import org.apache.spark.streaming.kafka010.KafkaUtils;
import org.apache.spark.streaming.kafka010.LocationStrategies;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.Map;
public class OmdSparkStreamService {
private static final Map<String, Object> KAFKA_CONSUMER_PROPERTIES;
static {
KAFKA_CONSUMER_PROPERTIES = Collections.unmodifiableMap(OmdKafkaConnection.kakfkaconfig());
}
private static final Collection<String> TOPICS =
Collections.unmodifiableList(Arrays.asList(OmdKafkaUtility.KAFKA_TOPIC));
public static void main(String[] args)throws InterruptedException {
JavaInputDStream<ConsumerRecord<String, String>> orderStream = null;
System.setProperty("hadoop.home.dir", OmdKafkaUtility.HADOOP_HOME_DIR_VALUE);
SparkConf conf = new SparkConf()
.setMaster(OmdKafkaUtility.RUN_LOCAL_WITH_AVAILABLE_CORES)
.setAppName(OmdKafkaUtility.APPLICATION_NAME)
.set("spark.mongodb.output.uri", OmdKafkaUtility.MONGODB_OUTPUT_URI);
JavaStreamingContext streamingContext
= new JavaStreamingContext(conf, new Duration(OmdKafkaUtility.BATCH_DURATION_INTERVAL_MS));
orderStream =
KafkaUtils.createDirectStream(
streamingContext,
LocationStrategies.PreferConsistent(),
ConsumerStrategies.<String, String>Subscribe(TOPICS, KAFKA_CONSUMER_PROPERTIES)
);
OmdSparkStreamRepositoryImpl.saveMongoData(orderStream);
streamingContext.start();
streamingContext.awaitTermination();
}
}
omd-stream-processor/src/main/resources/log4j.properties 0 → 100755
View file @ e1720051
log4j.rootLogger=WARN, console
log4j.logger.com.example=INFO
log4j.logger.org.apache=WARN
log4j.appender.console=org.apache.log4j.ConsoleAppender
log4j.appender.console.Target=System.out
log4j.appender.console.layout=org.apache.log4j.PatternLayout
log4j.appender.consoleAppender.layout.ConversionPattern=[%-5level] %d{dd-MM-yyyy hh:mm:ss.SSS a} [%thread] %logger{1} - %msg%n
\ No newline at end of file
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment