Skip to content

S
SparkPoc1
Commit 5b3fc20e authored by Sai Teja Ejjigiri's avatar Sai Teja Ejjigiri
Browse files
Options

Commit for reading the file from local and woth spark and writing it into the csv

parent 2402140f
Show whitespace changes
Inline Side-by-side
Showing with 43 additions and 108 deletions
pom.xml
View file @ 5b3fc20e
<?xml version="1.0" encoding="UTF-8"?> <?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" <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"> xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion> <modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.springframework.boot</groupId> <groupId>com.example</groupId>
<artifactId>spring-boot-starter-parent</artifactId> <artifactId>demo</artifactId>
<version>2.7.12</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>
<groupId>com.nisum</groupId>
<artifactId>sparkpoc</artifactId>
<version>0.0.1-SNAPSHOT</version> <version>0.0.1-SNAPSHOT</version>
<name>sparkpoc</name> <packaging>jar</packaging>
<description>Demo project for Spark Poc</description>
<name>demo</name>
<description>Demo project for Spring Boot</description>
<properties> <properties>
<java.version>1.8</java.version> <maven.compiler.source>8</maven.compiler.source>
<maven.compiler.target>8</maven.compiler.target>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<spark.version>3.2.1</spark.version>
</properties> </properties>
<dependencies> <dependencies>
<dependency> <dependency>
<groupId>org.springframework.boot</groupId> <groupId>org.apache.spark</groupId>
<artifactId>spring-boot-starter-web</artifactId> <artifactId>spark-sql_2.12</artifactId>
</dependency> <version>${spark.version}</version>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-streams</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
</dependency> </dependency>
<dependency> <dependency>
<groupId>org.apache.spark</groupId> <groupId>org.apache.spark</groupId>
<artifactId>spark-sql_2.12</artifactId> <artifactId>spark-core_2.12</artifactId>
<version>3.1.2</version> <version>${spark.version}</version>
</dependency> </dependency>
<dependency> <dependency>
<groupId>org.apache.spark</groupId> <groupId>org.apache.spark</groupId>
<artifactId>spark-streaming_2.12</artifactId> <artifactId>spark-streaming_2.12</artifactId>
<version>3.1.2</version> <version>${spark.version}</version>
</dependency> </dependency>
<!-- Logging dependency -->
<dependency> <dependency>
<groupId>ch.qos.logback</groupId> <groupId>org.apache.spark</groupId>
<artifactId>logback-classic</artifactId> <artifactId>spark-streaming-kafka-0-10_2.12</artifactId>
<version>${spark.version}</version>
</dependency> </dependency>
<!-- For Apache Spark -->
<dependency> <dependency>
<groupId>org.apache.spark</groupId> <groupId>org.apache.spark</groupId>
<artifactId>spark-core_2.12</artifactId> <artifactId>spark-sql-kafka-0-10_2.12</artifactId>
<version>3.2.0</version> <version>${spark.version}</version>
</dependency> </dependency>
<dependency> <dependency>
<groupId>org.projectlombok</groupId> <groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId> <artifactId>lombok</artifactId>
<optional>true</optional> <optional>true</optional>
</dependency> </dependency>
<dependency> <dependency>
<groupId>org.springframework.boot</groupId> <groupId>org.projectlombok</groupId>
<artifactId>spring-boot-starter-test</artifactId> <artifactId>lombok</artifactId>
<scope>test</scope> <version>1.18.26</version>
<scope>compile</scope>
</dependency> </dependency>
<dependency> <dependency>
<groupId>org.springframework.kafka</groupId> <groupId>com.opencsv</groupId>
<artifactId>spring-kafka-test</artifactId> <artifactId>opencsv</artifactId>
<scope>test</scope> <version>5.5.2</version>
</dependency> </dependency>
</dependencies>
<build> </dependencies>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
<configuration>
<excludes>
<exclude>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
</exclude>
</excludes>
</configuration>
</plugin>
</plugins>
</build>
</project> </project>
src/main/java/com/example/demo/DemoApplication.java deleted 100644 → 0
View file @ 2402140f
package com.example.demo;
import org.springframework.boot.*;
import org.springframework.boot.autoconfigure.*;
import org.springframework.web.bind.annotation.*;
@SpringBootApplication
@RestController
public class DemoApplication {
@GetMapping("/")
String home() {
return "Spring is here!";
}
public static void main(String[] args) {
SpringApplication.run(DemoApplication.class, args);
}
}
\ No newline at end of file
src/main/java/com/example/demo/SparkStreamingData.java
View file @ 5b3fc20e
...@@ -17,7 +17,7 @@ public class SparkStreamingData { ...@@ -17,7 +17,7 @@ public class SparkStreamingData {
Map<String, String> kafkaConfigMap = new HashMap<>(); Map<String, String> kafkaConfigMap = new HashMap<>();
kafkaConfigMap.put("kafka.bootstrap.servers", "localhost:9092"); kafkaConfigMap.put("kafka.bootstrap.servers", "localhost:9092");
kafkaConfigMap.put("subscribe", "MyStreamingTopic"); kafkaConfigMap.put("subscribe", "SparkOutput");
kafkaConfigMap.put("startingOffsets", "earliest"); kafkaConfigMap.put("startingOffsets", "earliest");
kafkaConfigMap.put("failOnDataLoss", "false"); kafkaConfigMap.put("failOnDataLoss", "false");
kafkaConfigMap.put("maxOffsetsPerTrigger","100"); kafkaConfigMap.put("maxOffsetsPerTrigger","100");
...@@ -71,7 +71,7 @@ public class SparkStreamingData { ...@@ -71,7 +71,7 @@ public class SparkStreamingData {
private static Dataset<Row> readKafkaCsvData(SparkSession sparkSession) { private static Dataset<Row> readKafkaCsvData(SparkSession sparkSession) {
Map<String, String> kafkaConfigData = new HashMap<>(); Map<String, String> kafkaConfigData = new HashMap<>();
kafkaConfigData.put("kafka.bootstrap.servers","127.0.0.1:9092"); kafkaConfigData.put("kafka.bootstrap.servers","127.0.0.1:9092");
kafkaConfigData.put("subscribe","MyStreamingTopic"); kafkaConfigData.put("subscribe","SparkOutput");
kafkaConfigData.put("startingOffsets","earliest"); kafkaConfigData.put("startingOffsets","earliest");
kafkaConfigData.put("failOnDataLoss","false"); kafkaConfigData.put("failOnDataLoss","false");
kafkaConfigData.put("kafka.con.group.id","localStream"); kafkaConfigData.put("kafka.con.group.id","localStream");
......
src/test/java/com/example/demo/DemoApplicationTests.java deleted 100644 → 0
View file @ 2402140f
package com.example.demo;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.boot.test.context.SpringBootTest.WebEnvironment;
import org.springframework.boot.test.web.client.TestRestTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.test.context.junit4.SpringRunner;
import static org.assertj.core.api.Assertions.assertThat;
@RunWith(SpringRunner.class)
@SpringBootTest(webEnvironment = WebEnvironment.RANDOM_PORT)
public class DemoApplicationTests {
@Test
public void contextLoads() {
}
@Autowired
private TestRestTemplate restTemplate;
@Test
public void homeResponse() {
String body = this.restTemplate.getForObject("/", String.class);
assertThat(body).isEqualTo("Spring is here!");
}
}
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