java框架在实时大数据处理中的应用

// 声明处理环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// 添加数据源
DataStream<SensorReading> readings = env.addSource(new FlinkKafkaProducer(KAFKA_TOPIC));

// 过滤数据，只保留温度高于阈值的读数
DataStream<SensorReading> filteredreadings = readings.filter(reading -> reading.temperature > TEMP_THRESHOLD);

// 计算每个传感器平均温度
DataStream<SensorAverage> averages = filteredreadings.keyBy("sensorId")
  .timeWindow(Time.seconds(600)) // 将数据流切分成 10 分钟的窗口
  .aggregate(new AverageAggregator());

// 打印结果
averages.print();

env.execute("IoT Data Processing");

// 声明处理环境
StreamingContext ssc = new StreamingContext(sparkConf, Seconds(60));

// 添加数据源
JavaDStream<String> tweets = ssc.socketTextStream(HOST, PORT);

// 检测负面情绪
JavaDStream<Long> negativeTweets = tweets.flatMap(tweet -> Arrays.asList(tweet.split(" "))
  .filter(word -> isNegative(word)));

// 累计负面情绪
JavaDStream<Long> runningTotal = negativeTweets.cumulativeCount();

// 打印结果
runningTotal.print();

ssc.start();
ssc.awaitTermination();