【SQL 周周练】爬取短视频发现数据缺失，如何用 SQL 填充

import numpy as np
from scipy import stats
import pandas as pd
import datetime

SEED = 2025
rng = np.random.default_rng(SEED)

# 开始日期
START_DATETIME = datetime.datetime(2025, 5, 1)
# 短视频数量
VIDEO_NUM = 100
# 一般最高点赞量
GENERAL_HIGHEST_LIKES_NUM = 1_000_000
# 一般最低点赞量
GENERAL_LOWEST_LIKES_NUM = 1

• 1.
• 2.
• 3.
• 4.
• 5.
• 6.
• 7.
• 8.
• 9.
• 10.
• 11.
• 12.
• 13.
• 14.
• 15.
• 16.

# 参考正态分布 3-sigma，转换到对数正态分布的范围
sigma = np.log(GENERAL_HIGHEST_LIKES_NUM / GENERAL_LOWEST_LIKES_NUM) / 6

# 如果在 JupyterLab 中分为不同单元格执行，
# 必须重置随机数生成器，否则不能复现同样结果
# rng = np.random.default_rng(SEED)
K_arr = stats.lognorm.rvs(
    s=sigma,
    loc=0,
    scale=np.sqrt(GENERAL_HIGHEST_LIKES_NUM * GENERAL_LOWEST_LIKES_NUM),
    size=VIDEO_NUM,
    random_state=rng,
)
K_arr = np.round(K_arr, 0)

# 参数 a 随机生成
a_arr = rng.uniform(low=0.85, high=0.95, size=VIDEO_NUM)
# 参数 b 随机生成
b_arr = rng.uniform(low=0.75, high=0.85, size=VIDEO_NUM)
# 参数 t0 随机生成
t0_arr = rng.choice([7,8,9,10,11,12], size=VIDEO_NUM, replace=True, p=[0.05,0.1,0.25,0.35,0.2,0.05])

# 定义 Gompertz 函数，为什么不选常见的 Logistic 函数
# 纯粹为了多尝试尝试以前没玩过的
defgompertz_func(t, K=100000, a=0.9, b=0.8, t0=10):
    return K * np.power(a, np.power(b, t-t0))

• 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.

days = np.arange(1, 31)

# 定义随机缺失；1 - 表示缺失数据
random_missing_flag = rng.choice([0, 1], size=VIDEO_NUM * len(days), p=[0.7, 0.3])

video_data = { 'video_id': [], 'dt': [], 'likes_num': [], 'random_miss': random_missing_flag}

for i, (k, a, b, t0) inenumerate(zip(K_arr, a_arr, b_arr, t0_arr)):
    # 视频 id 数字部分不低于 3 位，补零
    video_data['video_id'].extend([f'video_{i:03}']*len(days))
    video_data['dt'].extend(days)
    video_data['likes_num'].extend(gompertz_func(days, k, a, b, t0))    

df = pd.DataFrame(video_data)

# 四舍五入并转为整型
df['likes_num'] = np.round(df['likes_num'], 0).astype(int)
df['dt'] = df['dt'].apply(lambda d: START_DATETIME + datetime.timedelta(days=(d-1)))
df['show_likes_num'] = df.apply(lambda r: np.nan if r['random_miss'] == 1else r['likes_num'], axis = 1)
df.drop(['random_miss'], axis=1, inplace=True)

out_csv_path = "dwd_short_videos_likes_num_missing_data_from_crawler.csv"
df.to_csv(out_csv_path, header=False, index=False, encoding='utf-8-sig')

• 1.
• 2.
• 3.
• 4.
• 5.
• 6.
• 7.
• 8.
• 9.
• 10.
• 11.
• 12.
• 13.
• 14.
• 15.
• 16.
• 17.
• 18.
• 19.
• 20.
• 21.
• 22.
• 23.

from pyhive import hive

# 配置连接参数
host_ip = "127.0.0.1"
port = 10000
username = "蒋点数分"

with hive.Connection(host=host_ip, port=port) as conn:
    cursor = conn.cursor()
    
    hive_table_name = "data_exercise.dwd_short_videos_likes_num_missing_data_from_crawler"

    drop_table_sql = f"""
    drop table if exists {hive_table_name}
    """

    print(drop_table_sql)
    cursor.execute(drop_table_sql)

    create_table_sql = f"""
    create table if not exists {hive_table_name} (
        video_id string comment "短视频id",
        dt string comment "日期",
        likes_num int comment "点赞量(用来对比结果)",
        show_likes_num int comment "展示点赞量(用来补全数据)"
    ) 
    comment "短视频点赞量缺失数据，用来练习 SQL 补全数据 | author: 蒋点数分 | 文章编号：0a94d809"
    row format delimited fields terminated by ","
    stored as textfile
    """

    print(create_table_sql)
    cursor.execute(create_table_sql)

    import os

    load_data_sql = f"""
    load data local inpath "{os.path.abspath(out_csv_path)}"
    overwrite into table {hive_table_name}
    """

    print(load_data_sql)
    cursor.execute(load_data_sql)

    cursor.close()

• 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.

with calc_exist_likes_num as (
    select
       video_id, dt, likes_num, show_likes_num
    -- 找前面的非空点赞量
    -- 注意第二个参数为 true 表示跳过 null
    , last_value(show_likes_num, true) over(partitionby video_id orderby dt asc
            rowsbetween unbounded preceding and1 preceding) as last_exist_likes_num
    from data_exercise.dwd_short_videos_likes_num_missing_data_from_crawler
)

select
  video_id, dt, likes_num, show_likes_num
-- nvl 只支持 2 个参数；使用 coalesce 
-- 按照顺序，返回第一个非 Null 的值
-- 根据我写的兜底逻辑，如果向前寻找非空数值没有，则用 0 兜底
, coalesce(show_likes_num, last_exist_likes_num, 0) as forward_fill
from calc_exist_likes_num

• 1.
• 2.
• 3.
• 4.
• 5.
• 6.
• 7.
• 8.
• 9.
• 10.
• 11.
• 12.
• 13.
• 14.
• 15.
• 16.
• 17.

with calc_exist_likes_num as (
    select
       video_id, dt, likes_num, show_likes_num
    -- 前后邻近的第一个非空点赞量
    -- 注意第二个参数为 true，表示跳过 null
    , last_value(show_likes_num, true) over(partitionby video_id orderby dt asc
            rowsbetween unbounded preceding and1 preceding) as last_exist_likes_num
    
    , first_value(show_likes_num, true) over(partitionby video_id orderby dt asc
        rowsbetween1 following and unbounded following) as next_exist_likes_num
    from data_exercise.dwd_short_videos_likes_num_missing_data_from_crawler
)

select
  video_id, dt, likes_num, show_likes_num
-- nvl 只支持 2 个参数；使用 coalesce 
-- 按照顺序，返回第一个非 Null 的值
-- 根据我写的兜底逻辑，如果向后寻找非空数值没有，则用前面的第一个非空兜底，最后用 0 兜底
, coalesce(show_likes_num, next_exist_likes_num, last_exist_likes_num, 0) as backward_fill
from calc_exist_likes_num

• 1.
• 2.
• 3.
• 4.
• 5.
• 6.
• 7.
• 8.
• 9.
• 10.
• 11.
• 12.
• 13.
• 14.
• 15.
• 16.
• 17.
• 18.
• 19.
• 20.

with calc_exist_likes_num as (
    select
       video_id, dt, likes_num, show_likes_num
    -- 前后邻近的第一个非空点赞量
    -- 注意第二个参数为 true，表示跳过 null
    , last_value(show_likes_num, true) over(partitionby video_id orderby dt asc
            rowsbetween unbounded preceding and1 preceding) as last_exist_likes_num
    
    , first_value(show_likes_num, true) over(partitionby video_id orderby dt asc
        rowsbetween1 following and unbounded following) as next_exist_likes_num
    from data_exercise.dwd_short_videos_likes_num_missing_data_from_crawler
)

select
  video_id, dt, likes_num, show_likes_num
-- 如果一开始的数据没有，就按照零计算；因此“上一个点赞量必然存在”
-- 如果下一个点赞量不存在，那么就按照零计算，实际就等于不算
-- 注意四舍五入取整，我这里不取整，是为了给大家更方便的验证结果
, nvl(show_likes_num,  (nvl(last_exist_likes_num, 0) + nvl(next_exist_likes_num,0)) 
    / (1+ if(next_exist_likes_num isnot null, 1, 0))
 ) as half_fill_likes_num
from calc_exist_likes_num

• 1.
• 2.
• 3.
• 4.
• 5.
• 6.
• 7.
• 8.
• 9.
• 10.
• 11.
• 12.
• 13.
• 14.
• 15.
• 16.
• 17.
• 18.
• 19.
• 20.
• 21.
• 22.

with calc_exist_likes_num as (
    select
      video_id, dt, likes_num, show_likes_num
    -- 注意将 first_value 和 last_value 的第二个参数设置为 true 表示跳过 null
    , last_value(show_likes_num, true) over(partitionby video_id 
            orderby dt rowsbetween unbounded preceding and1 preceding) as last_exist_likes_num
    , first_value(show_likes_num, true) over(partitionby video_id 
            orderby dt rowsbetween1 following and unbounded following) as next_exist_likes_num
    -- 取出前后存在的点赞量对应的日期端点
    -- 注意要根据点赞量同时将日期设为 null
    , last_value(if(show_likes_num isnull, null, dt), true) over(partitionby video_id 
            orderby dt rowsbetween unbounded preceding and1 preceding) as last_exist_likes_dt
    , first_value(if(show_likes_num isnull, null, dt), true) over(partitionby video_id 
            orderby dt rowsbetween1 following and unbounded following) as next_exist_likes_dt
    , min(dt) over(partitionby video_id) as base_dt
    from data_exercise.dwd_short_videos_likes_num_missing_data_from_crawler
)

select
  video_id, dt, likes_num, show_likes_num
-- 推导见说明，此处不四舍五入也是为了验证结果
, nvl(show_likes_num, if( e isnull, s, (s * n_i + e * i_m) / (n_i + i_m))) as percentile_fill_likes_num
from (
    select
      video_id, dt, likes_num, show_likes_num
    , nvl(last_exist_likes_num,0) as s
    , next_exist_likes_num as e
    , if(last_exist_likes_num isnot null, datediff(dt, last_exist_likes_dt), datediff(dt, base_dt)+1) as i_m
    , datediff(next_exist_likes_dt, dt) as n_i
    from calc_exist_likes_num
) t
-- 经过一通处理，发现原来的顺序被破坏，重新排序
orderby video_id asc, dt asc

• 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.