1
// Mapper：输入 (行号, 文本) → 输出 (单词, 1)
2
public class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
3
    private Text word = new Text();
4
    @Override
5
    protected void map(LongWritable key, Text value, Context context)
6
            throws IOException, InterruptedException {
7
        for (String token : value.toString().split("\\s+")) {
8
            word.set(token);
9
            context.write(word, new IntWritable(1));
10
        }
11
    }
12
}
13

14
// Reducer：输入 (单词, [1,1,...]) → 输出 (单词, 总数)
15
public class WordCountReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
16
    @Override
17
    protected void reduce(Text key, Iterable<IntWritable> values, Context context)
18
            throws IOException, InterruptedException {
19
        int sum = 0;
20
        for (IntWritable val : values) { sum += val.get(); }
21
        context.write(key, new IntWritable(sum));
22
    }
23
}

1
from pyspark.sql import SparkSession
2
from pyspark.sql import functions as F
3

4
spark = SparkSession.builder \
5
    .appName("UserAnalytics") \
6
    .config("spark.executor.memory", "4g") \
7
    .getOrCreate()
8

9
# 读取数据（惰性求值，不立即执行）
10
orders = spark.read.parquet("hdfs:///data/orders/")
11
users = spark.read.parquet("hdfs:///data/users/")
12

13
# 转换操作（构建 DAG，仍然惰性）
14
result = (
15
    orders
16
    .join(users, orders.user_id == users.id)       # Join
17
    .filter(F.col("status") == "completed")         # 过滤
18
    .groupBy("city")                                # 分组
19
    .agg(
20
        F.sum("amount").alias("total_amount"),
21
        F.count("*").alias("order_count"),
22
        F.avg("amount").alias("avg_amount")
23
    )
24
    .orderBy(F.desc("total_amount"))                 # 排序
25
)
26

27
# 行动操作（触发实际计算）
28
result.show(10)
29
result.write.mode("overwrite").parquet("hdfs:///output/city_stats/")

1
// Flink 实时订单统计
2
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
3

4
KafkaSource<Order> source = KafkaSource.<Order>builder()
5
    .setBootstrapServers("kafka:9092")
6
    .setTopics("orders")
7
    .setGroupId("order-stats")
8
    .setStartingOffsets(OffsetsInitializer.earliest())
9
    .setValueOnlyDeserializer(new JsonDeserializationSchema<>(Order.class))
10
    .build();
11

12
DataStream<Order> orders = env.fromSource(
13
    source, WatermarkStrategy.<Order>forBoundedOutOfOrderness(Duration.ofSeconds(5))
14
        .withTimestampAssigner((event, ts) -> event.getTimestamp()),
15
    "Kafka Source"
16
);
17

18
// 窗口聚合：每5分钟统计各城市订单金额
19
orders.filter(o -> o.getStatus().equals("completed"))
20
    .keyBy(Order::getCity)
21
    .window(TumblingEventTimeWindows.of(Time.minutes(5)))
22
    .aggregate(new OrderAggregator())
23
    .addSink(new ElasticsearchSink<>());
24

25
env.execute("Real-time Order Analytics");

1
// Watermark 策略：允许5秒的乱序
2
WatermarkStrategy<Order> watermarkStrategy = WatermarkStrategy
3
    .<Order>forBoundedOutOfOrderness(Duration.ofSeconds(5))
4
    .withTimestampAssigner((order, recordTimestamp) -> order.getEventTime())
5
    .withIdleness(Duration.ofMinutes(1));  // 空闲分区处理
6

7
// 延迟数据处理：侧输出
8
OutputTag<Order> lateDataTag = new OutputTag<Order>("late-data") {};
9

10
SingleOutputStreamOperator<CityStats> result = orders
11
    .keyBy(Order::getCity)
12
    .window(TumblingEventTimeWindows.of(Time.minutes(5)))
13
    .allowedLateness(Time.minutes(1))  // 允许1分钟延迟
14
    .sideOutputLateData(lateDataTag)   // 超出延迟的数据进入侧输出
15
    .aggregate(new OrderAggregator());
16

17
// 单独处理延迟数据
18
DataStream<Order> lateOrders = result.getSideOutput(lateDataTag);
19
lateOrders.addSink(new AlertSink());  // 告警或写入冷存储

1
# Flink Checkpoint 配置
2
env.enable_checkpointing(60000)  # 每60秒一次Checkpoint
3
env.get_checkpoint_config().set_checkpointing_mode(CheckpointingMode.EXACTLY_ONCE)
4
env.get_checkpoint_config().set_checkpoint_timeout(120000)  # 超时2分钟
5
env.get_checkpoint_config().set_min_pause_between_checkpoints(30000)  # 最小间隔30秒
6
env.get_checkpoint_config().set_max_concurrent_checkpoints(1)  # 最多1个并发Checkpoint
7
env.get_checkpoint_config().set_tolerable_checkpoint_failure_number(3)  # 允许3次失败

1
// Debezium MySQL Connector 核心配置
2
{
3
  "name": "inventory-connector",
4
  "config": {
5
    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
6
    "database.hostname": "mysql",
7
    "database.port": "3306",
8
    "database.user": "debezium",
9
    "database.password": "dbz",
10
    "database.server.name": "dbserver1",
11
    "database.include.list": "inventory",
12
    "table.include.list": "inventory.orders,inventory.users",
13
    "snapshot.mode": "schema_only"
14
  }
15
}

1
// 传统 CRUD：只保存最终状态，旧状态丢失
2
public class OrderService {
3
    public void updateOrderStatus(Long orderId, String status) {
4
        Order order = orderRepository.findById(orderId);
5
        order.setStatus(status);
6
        orderRepository.save(order);
7
    }
8
}
9

10
// Event Sourcing：保存事件序列，状态由回放派生
11
public class EventSourcedOrderService {
12
    public void handle(CreateOrderCommand cmd) {
13
        eventStore.append(cmd.getOrderId(), List.of(
14
            new OrderCreatedEvent(cmd.getOrderId(), cmd.getItems(), cmd.getTotal())));
15
    }
16
    public void handle(PayOrderCommand cmd) {
17
        OrderState state = eventStore.replay(cmd.getOrderId());
18
        if (state.getStatus() != Status.CREATED) throw new IllegalStateException("订单已支付");
19
        eventStore.append(cmd.getOrderId(), List.of(
20
            new OrderPaidEvent(cmd.getOrderId(), cmd.getAmount(), cmd.getMethod())));
21
    }
22
}

1
# CQRS 读写分离示例：查询侧从事件流构建读模型
2
class OrderReadModel:
3
    def __init__(self, redis_client):
4
        self.redis = redis_client
5

6
    def handle_event(self, event):
7
        """消费事件，更新 Redis 读模型"""
8
        key = f"order:{event['order_id']}"
9
        if event["type"] == "OrderCreated":
10
            self.redis.hset(key, mapping={"status": "created", "total": event["total"]})
11
        elif event["type"] == "OrderPaid":
12
            self.redis.hset(key, mapping={"status": "paid", "paid_at": event["timestamp"]})
13
        elif event["type"] == "OrderShipped":
14
            self.redis.hset(key, mapping={"status": "shipped", "tracking": event["tracking"]})
15

16
    def get_order(self, order_id):
17
        return self.redis.hgetall(f"order:{order_id}")

1
// Kafka Streams：流表二象性的实际应用
2
StreamsBuilder builder = new StreamsBuilder();
3

4
// KStream：每条记录都是独立事件（点击流）
5
KStream<String, ClickEvent> clicks = builder.stream("click-events");
6

7
// KTable：每个 key 只保留最新值（用户信息）
8
KTable<String, UserProfile> profiles = builder.table("user-profiles");
9

10
// 流表 Join：点击事件关联用户信息
11
KStream<String, EnrichedClick> enriched = clicks.leftJoin(
12
    profiles,
13
    (click, profile) -> new EnrichedClick(click, profile)
14
);
15

16
// 流聚合为表：按用户统计点击量
17
KTable<String, Long> clickCounts = clicks
18
    .groupBy((key, click) -> click.getUserId())
19
    .count(Materialized.as("click-counts"));
20

21
// 表变更转为流：点击量变化事件
22
KStream<String, Long> countChanges = clickCounts.toStream();

1
-- Flink SQL：同一套语法，批流统一
2
CREATE TABLE orders (
3
    order_id BIGINT,
4
    user_id BIGINT,
5
    amount DECIMAL(10, 2),
6
    status STRING,
7
    event_time TIMESTAMP(3),
8
    WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND
9
) WITH (
10
    'connector' = 'kafka',
11
    'topic' = 'orders',
12
    'properties.bootstrap.servers' = 'kafka:9092',
13
    'format' = 'json'
14
);
15

16
-- 连续查询：每5分钟统计订单金额
17
SELECT
18
    city,
19
    TUMBLE_START(event_time, INTERVAL '5' MINUTE) AS window_start,
20
    TUMBLE_END(event_time, INTERVAL '5' MINUTE) AS window_end,
21
    COUNT(*) AS order_count,
22
    SUM(amount) AS total_amount
23
FROM orders
24
GROUP BY city, TUMBLE(event_time, INTERVAL '5' MINUTE);