锦中融合门户系统

package com.example.serviceportal;

import org.springframework.boot.SpringApplication;

import org.springframework.boot.autoconfigure.SpringBootApplication;

import org.springframework.web.bind.annotation.GetMapping;

import org.springframework.web.bind.annotation.RestController;

@SpringBootApplication

public class ServicePortalApplication {

public static void main(String[] args) {

SpringApplication.run(ServicePortalApplication.class, args);

}

@RestController

class ServiceController {

@GetMapping("/services")

public String getServices() {

return "Available Services: Login, Registration, Help Desk";

}

}

}

package com.example.kafka;

import org.apache.kafka.clients.producer.*;

import java.util.Properties;

public class ServiceLogProducer {

public static void main(String[] args) {

Properties props = new Properties();

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

props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");

props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

Producer producer = new KafkaProducer<>(props);

for (int i = 0; i < 10; i++) {

String logMessage = "Service call: /services, Timestamp: " + System.currentTimeMillis();

ProducerRecord record = new ProducerRecord<>("service_logs", logMessage);

producer.send(record);

}

producer.close();

}

}

package com.example.spark;

import org.apache.spark.SparkConf;

import org.apache.spark.streaming.Durations;

import org.apache.spark.streaming.api.java.JavaStreamingContext;

import java.util.Arrays;

import java.util.Set;

public class ServiceLogConsumer {

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

SparkConf conf = new SparkConf().setAppName("ServiceLogConsumer").setMaster("local[*]");

JavaStreamingContext jssc = new JavaStreamingContext(conf, Durations.seconds(5));

Set topics = Set.of("service_logs");

jssc.createDirectStream(

"localhost:9092",

"service_logs",

"group1"

).map(record -> record.value())

.foreachRDD(rdd -> rdd.foreach(line -> System.out.println("Received log: " + line)));

jssc.start();

jssc.awaitTermination();

}

}

package com.example.spark;

import org.apache.spark.sql.Dataset;

import org.apache.spark.sql.Row;

import org.apache.spark.sql.SparkSession;

public class ServiceCallCounter {

public static void main(String[] args) {

SparkSession spark = SparkSession.builder()

.appName("ServiceCallCounter")

.getOrCreate();

Dataset logs = spark.read().json("path/to/logs.json");

logs.createOrReplaceTempView("logs");

Dataset serviceCounts = spark.sql(

"SELECT service, COUNT(*) AS count FROM logs GROUP BY service"

);

serviceCounts.show();

spark.stop();

}

}