Java大数据处理框架的案例研究

// 创建 Spark StreamingContext
SparkConf conf = new SparkConf().setAppName("StreamingExample");
JavaStreamingContext jsc = new JavaStreamingContext(conf, Durations.seconds(5));

// 定义从 Kafka 接收数据的 DataStream
JavaDStream<String> lines = jsc.socketTextStream("localhost", 9999);

// 处理数据，检测并预测设备故障
JavaDStream<String> alerts = lines.flatMap(new FlatMapFunction<String, String>() {
   public Iterator<String> call(String line) {
       // 分割数据并检测故障
       String[] parts = line.split(",");
       if (Integer.parseInt(parts[1]) > 100) {
           return Arrays.asList("故障：设备 " + parts[0]).iterator();
       }
       return Collections.emptyIterator();
   }
});

// 聚合告警并输出到控制台
alerts.foreachRDD(new Function<JavaRDD<String>, Void>() {
   public Void call(JavaRDD<String> rdd) {
       rdd.foreach(System.out::println);
       return null;
   }
});

// 启动流处理
jsc.start();
jsc.awaitTermination();

// 编写 Mapper 类
public class LogMapper implements Mapper<LongWritable, Text, Text, IntWritable> {

    public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String[] parts = value.toString().split(",");
        context.write(new Text(parts[0]), new IntWritable(1));
    }
}

// 编写 Reducer 类
public class LogReducer implements Reducer<Text, IntWritable, Text, IntWritable> {

    public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
        int sum = 0;
        for (IntWritable value : values) {
            sum += value.get();
        }
        context.write(key, new IntWritable(sum));
    }
}

// 配置 Hadoop 作业
Configuration conf = new Configuration();
conf.set("mapred.job.name", "LogAnalysis");
conf.set("mapred.input.dir", "/input");
conf.set("mapred.output.dir", "/output");

// 提交作业
Job job = Job.getInstance(conf, "LogAnalysis");
job.setJarByClass(LogAnalysis.class);
job.setMapperClass(LogMapper.class);
job.setReducerClass(LogReducer.class);
job.setInputFormatClass(TextInputFormat.class);
job.setOutputFormatClass(TextOutputFormat.class);
job.waitForCompletion(true);