import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.types.StructType;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.functions;

public class NetworkTrafficAnalysis {

  public static void main(String[] args) {
    // 创建 SparkSession
    SparkSession spark = SparkSession.builder()
                  .master("local")
                  .appName("NetworkTrafficAnalysis")
                  .getOrCreate();

    // 定义输入文件的模式
    StructType schema = DataTypes.createStructType()
                  .add("timestamp", DataTypes.StringType)
                  .add("source_ip", DataTypes.StringType)
                  .add("destination_ip", DataTypes.StringType)
                  .add("bytes", DataTypes.LongType);

    // 读取数据文件
    Dataset<Row> trafficData = spark.read()
                  .schema(schema)
                  .csv("hdfs://<path_to_traffic_data>/network-traffic.csv");

    // 清理数据(例如，删除空行或无效行)
    trafficData = trafficData.filter(functions.col("timestamp").isNotNull());

    // 分组并聚合数据计算每小时的流量
    Dataset<Row> aggregatedTraffic = trafficData.groupBy(functions.window(functions.col("timestamp"), "1 hour"))
                   .agg(functions.sum("bytes").as("total_bytes"),
                         functions.min("source_ip").as("min_source_ip"),
                         functions.max("destination_ip").as("max_destination_ip"));

    // 从存储结果到持久存储(例如，HDFS 或数据库）
    aggregatedTraffic.write()
                   .format("csv")
                   .save("hdfs://<path_to_result>/aggregated-network-traffic.csv");
  }
}