字节面试攻略：真题解答与OA面经

				
					# 假设我们有一个用户“喜欢”视频的Schema
class LikeEvent:
    user_id: str
    video_id: str
    publish_time: int # Unix timestamp
    creator_id: str

# Flink DataStream API 伪代码
def process_tiktok_likes(env):
    # 1. 从Kafka消费用户“喜欢”视频的事件流
    like_events_stream = env.add_source(
        KafkaSource.builder()
        .setTopics("tiktok_like_events")
        .setStartingOffsets(OffsetsInitializer.earliest())
        .setValueOnlyDeserializer(LikeEventDeserializationSchema())
        .build()
    )

    # 2. 分配事件时间戳和Watermark
    # 使用视频的publish_time作为事件时间
    # 允许一定的乱序（例如5秒）
    watermarked_stream = like_events_stream.assignTimestampsAndWatermarks(
        WatermarkStrategy
        .forBoundedOutOfOrderness(Duration.ofSeconds(5))
        .withTimestampAssigner((event, timestamp) -> event.publish_time * 1000) # 转换为毫秒
    )

    # 3. 按创作者ID分组
    keyed_stream = watermarked_stream.keyBy(lambda event: event.creator_id)

    # 4. 定义滑动时间窗口 (24小时窗口，1分钟滑动)
    windowed_stream = keyed_stream.window(
        SlidingEventTimeWindows.of(Time.hours(24), Time.minutes(1))
    )

    # 5. 在窗口内聚合：计算每个创作者的视频数量
    # 这里使用ProcessWindowFunction，可以更灵活地处理状态和上下文信息
    processed_stream = windowed_stream.process(
        ProcessWindowFunction<LikeEvent, Tuple2, String, TimeWindow>() {
            private ValueState creatorVideoCount;

            @Override
            public void open(Configuration parameters) throws Exception {
                creatorVideoCount = getRuntimeContext().getState(
                    new ValueStateDescriptor("creatorVideoCount", Long.class)
                );
            }

            @Override
            public void process(String creatorId, Context context, Iterable elements, Collector<Tuple2> out) throws Exception {
                long currentCount = 0;
                if (creatorVideoCount.value() != null) {
                    currentCount = creatorVideoCount.value();
                }

                for (LikeEvent event : elements) {
                    currentCount++;
                }
                creatorVideoCount.update(currentCount);

                // 输出结果：(创作者ID, 24小时内被喜欢的视频数量)
                out.collect(new Tuple2(creatorId, currentCount));
            }
        }
    )

    # 6. 将结果输出到下游（例如另一个Kafka主题）
    processed_stream.add_sink(
        KafkaSink.<Tuple2>builder()
        .setBootstrapServers("kafka_broker:9092")
        .setRecordSerializer(
            KafkaRecordSerializationSchema.builder()
            .setTopic("tiktok_creator_video_counts")
            .setValueSerializationSchema(SimpleStringSchema()) # 或自定义序列化
            .build()
        )
        .build()
    )

    env.execute("TikTok Creator Video Count Job")

# 主函数
if __name__ == "__main__":
    env = StreamExecutionEnvironment.getExecutionEnvironment()
    # 根据实际情况配置并行度
    env.setParallelism(4)
    process_tiktok_likes(env)