大数据处理框架深度对比：Spark vs Flink vs Storm

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

// Spark核心概念
import org.apache.spark.sql.{SparkSession, DataFrame}
import org.apache.spark.sql.functions._

object SparkWordCount {
  def main(args: Array[String]): Unit = {
    // 创建SparkSession
    val spark = SparkSession.builder()
      .appName("WordCount")
      .master("local[*]")
      .getOrCreate()

    // 读取数据
    val textFile = spark.read.textFile("hdfs://path/to/input")

    // 转换操作
    val counts = textFile
      .rdd
      .flatMap(line => line.split(" "))
      .map(word => (word, 1))
      .reduceByKey(_ + _)

    // 行动操作
    counts.saveAsTextFile("hdfs://path/to/output")

    spark.stop()
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

// 结构化API
class SparkDataFrameExample(spark: SparkSession) {

  def processUserData(): Unit = {
    // 读取JSON数据
    val usersDF = spark.read
      .json("hdfs://data/users.json")

    // 注册临时视图
    usersDF.createOrReplaceTempView("users")

    // SQL查询
    val result = spark.sql("""
      SELECT
        department,
        COUNT(*) as user_count,
        AVG(age) as avg_age,
        SUM(salary) as total_salary
      FROM users
      WHERE status = 'active'
      GROUP BY department
      HAVING COUNT(*) > 10
      ORDER BY total_salary DESC
    """)

    result.show()

    // DSL风格
    val alternative = usersDF
      .filter(col("status") === "active")
      .groupBy("department")
      .agg(
        count("*").as("user_count"),
        avg("age").as("avg_age"),
        sum("salary").as("total_salary")
      )
      .filter(col("user_count") > 10)
      .orderBy(desc("total_salary"))
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

// 微批处理
import org.apache.spark.streaming._
import org.apache.spark.streaming.kafka._

class SparkStreamingExample {

  def createStreamingContext(): StreamingContext = {
    val conf = new SparkConf().setAppName("KafkaWordCount")
    val ssc = new StreamingContext(conf, Seconds(5)) // 5秒一批

    val kafkaParams = Map[String, Object](
      "bootstrap.servers" -> "localhost:9092",
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> "wordcount_group",
      "auto.offset.reset" -> "latest"
    )

    val topics = Array("input-topic")
    val stream = KafkaUtils.createDirectStream[
      String, String
    ](
      ssc,
      PreferConsistent,
      Subscribe[String, String](topics, kafkaParams)
    )

    val words = stream.flatMap(record =>
      record.value.split(" ")
    )

    val wordCounts = words.map(word => (word, 1L))
      .reduceByKey(_ + _)

    wordCounts.print()

    ssc
  }
}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101

// Flink DataStream API
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

public class FlinkStreamingExample {

    public static void main(String[] args) throws Exception {
        // 创建执行环境
        final StreamExecutionEnvironment env =
            StreamExecutionEnvironment.getExecutionEnvironment();

        // 设置并行度
        env.setParallelism(4);

        // 配置检查点（容错）
        env.enableCheckpointing(60000); // 60秒
        env.getCheckpointConfig().setCheckpointingMode(
            CheckpointingMode.EXACTLY_ONCE
        );

        // 创建数据源
        DataStream<SensorReading> readings = env
            .addSource(new SensorSource())
            .name("Sensor Source");

        // 数据处理
        DataStream<Alert> alerts = readings
            .keyBy(SensorReading::getId)
            .window(TumblingProcessingTimeWindows.of(Time.seconds(10)))
            .process(new AlertProcessor())
            .name("Alert Processor");

        // 输出
        alerts.addSink(new AlertSink());

        // 执行
        env.execute("Sensor Monitoring");
    }

    // 传感器数据源
    public static class SensorSource implements SourceFunction<SensorReading> {
        private volatile boolean isRunning = true;

        @Override
        public void run(SourceContext<SensorReading> ctx) throws Exception {
            Random random = new Random();

            while (isRunning) {
                SensorReading reading = new SensorReading(
                    "sensor-" + random.nextInt(10),
                    System.currentTimeMillis(),
                    random.nextDouble() * 100
                );
                ctx.collect(reading);
                Thread.sleep(100);
            }
        }

        @Override
        public void cancel() {
            isRunning = false;
        }
    }

    // 窗口处理函数
    public static class AlertProcessor extends
        ProcessWindowFunction<SensorReading, Alert, String, TimeWindow> {

        @Override
        public void process(
            String key,
            Context context,
            Iterable<SensorReading> readings,
            Collector<Alert> out
        ) {
            double sum = 0;
            int count = 0;

            for (SensorReading reading : readings) {
                sum += reading.getValue();
                count++;
            }

            double avg = sum / count;

            if (avg > 80) {
                Alert alert = new Alert(
                    key,
                    context.window().getEnd(),
                    avg
                );
                out.collect(alert);
            }
        }
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78

// Flink SQL
import org.apache.flink.table.api.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class FlinkTableExample {

    public static void main(String[] args) throws Exception {
        // 创建流环境
        StreamExecutionEnvironment env =
            StreamExecutionEnvironment.getExecutionEnvironment();

        StreamTableEnvironment tableEnv =
            StreamTableEnvironment.create(env);

        // 创建Kafka源表
        String kafkaSourceDDL = """
            CREATE TABLE orders (
                order_id BIGINT,
                user_id BIGINT,
                product_id INT,
                amount DECIMAL(10, 2),
                order_time TIMESTAMP(3),
                WATERMARK FOR order_time AS order_time - INTERVAL '5' SECOND
            ) WITH (
                'connector' = 'kafka',
                'topic' = 'orders',
                'properties.bootstrap.servers' = 'localhost:9092',
                'properties.group.id' = 'order-processor',
                'format' = 'json',
                'scan.startup.mode' = 'latest'
            )
        """;

        tableEnv.executeSql(kafkaSourceDDL);

        // 执行SQL查询
        String query = """
            SELECT
                product_id,
                TUMBLE_END(order_time, INTERVAL '1' HOUR) as window_end,
                COUNT(*) as order_count,
                SUM(amount) as total_amount
            FROM orders
            GROUP BY
                product_id,
                TUMBLE(order_time, INTERVAL '1' HOUR)
            HAVING COUNT(*) > 100
        """;

        Table result = tableEnv.sqlQuery(query);

        // 输出到控制台
        result.execute().print();

        // 或者写入Kafka
        /*
        String kafkaSinkDDL = """
            CREATE TABLE hourly_sales (
                product_id INT,
                window_end TIMESTAMP(3),
                order_count BIGINT,
                total_amount DECIMAL(10, 2)
            ) WITH (
                'connector' = 'kafka',
                'topic' = 'hourly-sales',
                'properties.bootstrap.servers' = 'localhost:9092',
                'format' = 'json'
            )
        """;

        tableEnv.executeSql(kafkaSinkDDL);
        result.executeInsert("hourly_sales");
        */

        env.execute("Hourly Sales Report");
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89

// Flink状态后端配置
import org.apache.flink.contrib.streaming.state.RocksDBStateBackend;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class FlinkStateBackend {

    public static void configureStateBackend(
        StreamExecutionEnvironment env
    ) throws Exception {
        // RocksDB状态后端（生产环境推荐）
        RocksDBStateBackend rocksDB = new RocksDBStateBackend(
            "hdfs://flink/checkpoints",  // 检查点目录
            true                          // 增量检查点
        );

        // 启用RocksDB
        env.setStateBackend(rocksDB);

        // 配置检查点
        env.enableCheckpointing(
            60000,                        // 间隔60秒
            CheckpointingMode.EXACTLY_ONCE, // 精确一次语义
            600000                         // 超时时间10分钟
        );

        // 设置并发检查点
        env.getCheckpointConfig().setConcurrentCheckpoints(1);

        // 设置最小检查点间隔
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(30000);

        // Keyed State示例
        DataStream<SensorReading> readings = env
            .addSource(new SensorSource());

        DataStream<Alert> alerts = readings
            .keyBy(SensorReading::getId)
            .process(new KeyedProcessFunction<String,
                SensorReading, Alert>() {

                // ValueState
                private ValueState<Double> lastValue;

                // ListState
                private ListState<Double> history;

                @Override
                public void open(Configuration parameters) {
                    ValueStateDescriptor<Double> lastValueDesc =
                        new ValueStateDescriptor<>(
                            "lastValue",
                            Double.class
                        );
                    lastValue = getRuntimeContext().getState(lastValueDesc);

                    ListStateDescriptor<Double> historyDesc =
                        new ListStateDescriptor<>(
                            "history",
                            Double.class
                        );
                    history = getRuntimeContext().getState(historyDesc);
                }

                @Override
                public void processElement(
                    SensorReading reading,
                    Context ctx,
                    Collector<Alert> out
                ) throws Exception {
                    // 获取上次的值
                    Double last = lastValue.value();

                    // 检查是否异常
                    if (last != null &&
                        Math.abs(reading.getValue() - last) > 50) {
                        out.collect(new Alert(
                            reading.getId(),
                            ctx.timestamp(),
                            "Value spike detected"
                        ));
                    }

                    // 更新状态
                    lastValue.update(reading.getValue());
                    history.add(reading.getValue());
                }
            });
    }
}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133

// Storm拓扑
import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.StormSubmitter;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichSpout;
import org.apache.storm.topology.base.BaseBasicBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Values;
import org.apache.storm.tuple.Tuple;

public class WordCountTopology {

    public static void main(String[] args) throws Exception {
        TopologyBuilder builder = new TopologyBuilder();

        // Spout: 数据源
        builder.setSpout("sentence-spout", new SentenceSpout(), 2);

        // Bolt: 分词
        builder.setBolt("split-bolt", new SplitBolt(), 4)
            .setNumTasks(8)
            .shuffleGrouping("sentence-spout");

        // Bolt: 计数
        builder.setBolt("count-bolt", new CountBolt(), 4)
            .fieldsGrouping("split-bolt", new Fields("word"));

        // Bolt: 报告
        builder.setBolt("report-bolt", new ReportBolt(), 1)
            .globalGrouping("count-bolt");

        Config config = new Config();
        config.setDebug(true);

        if (args != null && args.length > 0) {
            // 生产集群
            config.setNumWorkers(4);
            StormSubmitter.submitTopologyWithProgressBar(
                args[0],
                config,
                builder.createTopology()
            );
        } else {
            // 本地测试
            LocalCluster cluster = new LocalCluster();
            cluster.submitTopology("word-count", config,
                builder.createTopology());

            Thread.sleep(60000);
            cluster.shutdown();
        }
    }

    // Sentence Spout
    public static class SentenceSpout extends BaseRichSpout {
        private SpoutOutputCollector collector;
        private String[] sentences = {
            "the quick brown fox",
            "jumps over the lazy dog",
            "hello world from storm"
        };

        @Override
        public void open(
            Map conf,
            TopologyContext context,
            SpoutOutputCollector collector
        ) {
            this.collector = collector;
        }

        @Override
        public void nextTuple() {
            for (String sentence : sentences) {
                collector.emit(new Values(sentence));
            }
            Utils.sleep(100);
        }

        @Override
        public void declareOutputFields(
            OutputFieldsDeclarer declarer
        ) {
            declarer.declare(new Fields("sentence"));
        }
    }

    // Split Bolt
    public static class SplitBolt extends BaseBasicBolt {
        @Override
        public void execute(Tuple tuple, BasicOutputCollector collector) {
            String sentence = tuple.getStringByField("sentence");
            String[] words = sentence.split(" ");

            for (String word : words) {
                collector.emit(new Values(word));
            }
        }

        @Override
        public void declareOutputFields(
            OutputFieldsDeclarer declarer
        ) {
            declarer.declare(new Fields("word"));
        }
    }

    // Count Bolt
    public static class CountBolt extends BaseBasicBolt {
        private Map<String, Integer> counts = new HashMap<>();

        @Override
        public void execute(Tuple tuple, BasicOutputCollector collector) {
            String word = tuple.getStringByField("word");
            Integer count = counts.getOrDefault(word, 0) + 1;
            counts.put(word, count);

            collector.emit(new Values(word, count));
        }

        @Override
        public void declareOutputFields(
            OutputFieldsDeclarer declarer
        ) {
            declarer.declare(new Fields("word", "count"));
        }
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47

// 综合使用Spark和Flink
class RealTimeAnalytics {

  // Spark批量处理历史数据
  def processHistoricalData(): Unit = {
    val spark = SparkSession.builder()
      .appName("HistoricalAnalysis")
      .getOrCreate()

    val df = spark.read
      .parquet("hdfs://data/user-behavior/")

    val insights = df
      .groupBy("user_id")
      .agg(
        count("*").as("action_count"),
        collect_list("page_url").as("page_history"),
        approx_count_distinct("session_id").as("session_count")
      )

    insights.write
      .mode("overwrite")
      .parquet("hdfs://results/user-insights/")
  }

  // Flink实时处理新数据
  def processRealtimeData(): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment()

    val kafkaSource = new FlinkKafkaConsumer(
      "user-events",
      new JSONSchema,
      properties
    )

    val events = env.addSource(kafkaSource)

    val insights = events
      .keyBy(_.userId)
      .window(TumblingEventTimeWindows.of(Time.minutes(5)))
      .aggregate(new UserBehaviorAggregator())

    insights.addSink(new RedisSink())

    env.execute("RealtimeUserBehavior")
  }
}