大家好，我是“蔣點(diǎn)數(shù)分”，多年以來一直從事數(shù)據(jù)分析工作。從今天開始，與大家持續(xù)分享關(guān)于數(shù)據(jù)分析的學(xué)習(xí)內(nèi)容。

本文是第一篇，也是【SQL 周周練】系列的第一篇。該系列是挑選或自編具有一些難度的 SQL 題目，一周至少更新一篇。后續(xù)創(chuàng)作的內(nèi)容，初步規(guī)劃的方向包括：

后續(xù)內(nèi)容規(guī)劃

1.利用 Streamlit 實(shí)現(xiàn) Hive 元數(shù)據(jù)展示、SQL 編輯器、 結(jié)合Docker 沙箱實(shí)現(xiàn)數(shù)據(jù)分析 Agent

2.時(shí)間序列異常識(shí)別、異動(dòng)歸因算法

3.留存率擬合、預(yù)測(cè)、建模

4.學(xué)習(xí) AB 實(shí)驗(yàn)、復(fù)雜實(shí)驗(yàn)設(shè)計(jì)等

5.自動(dòng)化機(jī)器學(xué)習(xí)、自動(dòng)化特征工程

6.因果推斷學(xué)習(xí)

7. .……

歡迎關(guān)注，一起學(xué)習(xí)（歡迎準(zhǔn)備從事數(shù)據(jù)分析崗位的學(xué)生關(guān)注 一起來當(dāng)“工具人”）

第 1 期題目

題目來源：Uber 面試真題

一、題目介紹

有一張表，記錄了乘客對(duì)于司機(jī)的評(píng)價(jià)，請(qǐng)找出每個(gè)星期當(dāng)中連續(xù)獲得 5 星好評(píng)最多的 driver_id。列名：driver_id、rating_time、ratings （原題乘客 id 對(duì)解答題目是冗余的，故此我在文中省略掉...）連續(xù) 5 星，中間出現(xiàn)任意一次非 5 星，則中斷。

二、題目思路

想要答題的同學(xué)，可以先思考答案??。

.……

.……

.……

我來談?wù)勎业乃悸?，“連續(xù)”問題是數(shù)據(jù)分析師在 SQL 筆試中的“老朋友”了。最常見的就是“連續(xù)登錄”問題，其大概思路是利用日期減去排序row_number()得到一個(gè)“基準(zhǔn)日期”用來作為分組標(biāo)識(shí)。這里沒有日期，不能生搬硬套。

我們思維變通一下，如果想將連續(xù)計(jì)數(shù)的記錄能夠放在同一個(gè)組里，那么這個(gè)分組標(biāo)識(shí)是關(guān)鍵。對(duì)于連續(xù) 5 星，它們的有什么共同點(diǎn)？是每一個(gè) 5 星評(píng)價(jià)前面有多少個(gè)非 5 星（1～4 星）的評(píng)價(jià)。為了方便理解，我繪制一個(gè)簡(jiǎn)易的說明圖：

只需要注意剔除每組開頭可能多出來的非 5 星評(píng)價(jià)，即可完成統(tǒng)計(jì)。下面，我用 NumPy 結(jié)合一些假設(shè)來生成模擬的數(shù)據(jù)集：

三、生成模擬數(shù)據(jù)

只關(guān)心 SQL 代碼的同學(xué)，可以跳轉(zhuǎn)到第四節(jié)（我在工作中使用 Hive 較多，因此采用 Hive 的語法）

為了簡(jiǎn)化模擬數(shù)據(jù)的難度，做如下假設(shè)：

1.假設(shè)用戶下車之后立即評(píng)價(jià)，評(píng)價(jià)時(shí)間取下車時(shí)間
2.司機(jī)等待訂單、接客送客加在一起的時(shí)間間隔，通過指數(shù)分布模擬
3.訂單的時(shí)間間隔，不引入早晚高峰因素，不引入差異化因素 => 對(duì)每名司機(jī)的參數(shù)是一樣的
4.司機(jī)回家和睡覺的時(shí)間，算在一起，用正態(tài)分布模擬
5.不引入司機(jī)吃飯、出車前休息等個(gè)人事務(wù)的時(shí)間，否則模擬起來太復(fù)雜
6.對(duì)于司機(jī)，只限制每日最多在線時(shí)長(zhǎng)，不做周、月級(jí)別的限制
7.假設(shè)存在兩類司機(jī)：
	a.追求每天達(dá)到一個(gè)目標(biāo)收入，達(dá)到后則主動(dòng)收車 => 用單量代替收入
    b.追求每天達(dá)到某個(gè)在線時(shí)長(zhǎng)，達(dá)到后則主動(dòng)收車
 8.模擬數(shù)據(jù)累計(jì)后，可能導(dǎo)致的司機(jī)日夜規(guī)律顛倒 => 違背現(xiàn)實(shí)情況，不作調(diào)整

模擬代碼如下：

1.定義模擬邏輯需要的常量：

import datetime
import numpy as np
import pandas as pd

# 設(shè)置隨機(jī)數(shù)種子
np.random.seed(2025)
# 模擬的司機(jī)數(shù)量
DRIVER_NUM = 100
# 追求單量的司機(jī)數(shù)量（不論追求單量還是追求在線時(shí)長(zhǎng)，都要額外受平臺(tái)在線時(shí)長(zhǎng)限制）
PURSUING_ORDER_DRIVER_NUM = 55
# 追求訂單的數(shù)量取值 (10 ~ 20 單，值太高在其他參數(shù)影響下，也取不到)
# 離散均勻分布
pursuing_order_volume = np.random.choice(
    np.arange(10, 21), size=PURSUING_ORDER_DRIVER_NUM
)
# 追求在線時(shí)長(zhǎng)的司機(jī)數(shù)量
PURSUING_ONLINE_DRIVER_NUM = DRIVER_NUM - PURSUING_ORDER_DRIVER_NUM
# 追求在線時(shí)長(zhǎng)的取值 （8小時(shí)、8.5小時(shí)......12小時(shí)）
pursuing_online_duration = np.random.choice(
    np.arange(8, 12.5, 0.5), size=PURSUING_ONLINE_DRIVER_NUM
)
# 模擬數(shù)據(jù)的日期范圍
START_DATETIME = datetime.datetime(2025, 1, 1, 8, 0, 0)
END_DATETIME = datetime.datetime(2025, 5, 1, 23, 59, 59)
# 平均訂單時(shí)間間隔（單位秒，包含等單+接客+送客，等于評(píng)價(jià)時(shí)間間隔）
ORDER_INTERVAL_AVG = 40 * 60
# 司機(jī)平均休息時(shí)長(zhǎng)（單位秒，包含收車時(shí)間）
DRIVER_REST_DURATION_AVG = 8 * 3600
# 司機(jī)平均休息時(shí)長(zhǎng)標(biāo)準(zhǔn)差（單位秒）
DRIVER_REST_DURATION_STD = 30 * 60
# 每日在線時(shí)長(zhǎng)上限（秒）
ONLINE_DURATION_UPPER_LIMIT = 12 * 3600

2.模擬訂單間隔、乘客評(píng)分、休息間隔。為了提高生成速度，盡量一次讓 NumPy 生成足夠多的數(shù)據(jù)；用函數(shù)封裝起來，如果超出了預(yù)先生成的數(shù)據(jù)長(zhǎng)度，則開啟單次生成：

# 為了一次盡可能將數(shù)據(jù)模擬全
# 根據(jù)參數(shù)平均值，來計(jì)算出大概需要模擬出多少個(gè)訂單間隔，再增加 10% 浮動(dòng)
# round 函數(shù)輸出 float 類型，需要轉(zhuǎn)為 int 類型，不然后續(xù) numpy 的 size 會(huì)報(bào)錯(cuò)
ORDER_NUM_NEED_SIMULATION = int(
    round(
        (END_DATETIME - START_DATETIME).days
        * (ONLINE_DURATION_UPPER_LIMIT / ORDER_INTERVAL_AVG)
        * (1 + 0.1),
        0,
    )
)

# 生成模擬的訂單間隔
order_interval_simulation = np.random.exponential(
    scale=ORDER_INTERVAL_AVG, size=(DRIVER_NUM, ORDER_NUM_NEED_SIMULATION)
)

# 乘客的評(píng)價(jià)也一并隨機(jī)生成
rating_simulation = np.random.choice(
    np.arange(1, 6),
    size=(DRIVER_NUM, ORDER_NUM_NEED_SIMULATION),
    p=[0.01, 0.01, 0.02, 0.06, 0.9],
)


def get_order_interval_and_rating_simulation(driver_id, cnt):
    """
    獲取訂單間隔時(shí)長(zhǎng)和訂單評(píng)分，增加一個(gè)函數(shù)，
    是為了如果批量隨機(jī)生成的數(shù)據(jù)不夠用，再單次生成
    """
    if cnt >= ORDER_NUM_NEED_SIMULATION:
        return (
            np.random.exponential(scale=ORDER_INTERVAL_AVG),
            np.random.choice(np.arange(1, 6), p=[0.01, 0.01, 0.02, 0.06, 0.9]),
        )
    else:
        return (
            order_interval_simulation[driver_id][cnt],
            rating_simulation[driver_id][cnt],
        )


# 模擬休息的數(shù)據(jù)( 在線加休息的和有可能小于 24 小時(shí) )
REST_NUM_NEED_SIMULATION = int(
    round((END_DATETIME - START_DATETIME).days * (1 + 0.1), 0)
)

rest_interval_simulation = (
    np.clip(
        np.random.normal(loc=8, scale=0.5, size=(DRIVER_NUM, REST_NUM_NEED_SIMULATION)),
        a_min=6,
        a_max=12,
    )
    * 3600
)


def get_rest_interval_simulation(driver_id, cnt):
    """
    獲取休息間隔時(shí)長(zhǎng)，增加一個(gè)函數(shù)，是為了如果批量隨機(jī)生成的
    數(shù)據(jù)不夠用，再單次生成
    """
    if cnt >= REST_NUM_NEED_SIMULATION:
        return np.clip(np.random.normal(loc=8, scale=0.5), a_min=6, a_max=12) * 3600
    else:
        return rest_interval_simulation[driver_id][cnt]

2.根據(jù)假設(shè)的邏輯，生成司機(jī)的全部數(shù)據(jù)。注意司機(jī)休息的判斷條件，以及中間變量清零的處理：

table_data = {"driver_id": [], "rating_time": [], "ratings": []}

for driver_id in range(DRIVER_NUM):
    order_cnt = 0  # 第幾個(gè)訂單
    rest_cnt = 0  # 第幾次休息
    last_time = START_DATETIME
    # 當(dāng)前累計(jì)在線時(shí)間，注意單位是秒
    current_online_time = 0
    # 當(dāng)天的訂單，追求訂單的司機(jī)需要這個(gè)變量
    current_order_cnt = 0
    while True:
        table_data["driver_id"].append(driver_id)
        order_interval, rating = get_order_interval_and_rating_simulation(
            driver_id, order_cnt
        )
        last_time = last_time + datetime.timedelta(seconds=int(order_interval))
        table_data["rating_time"].append(last_time)
        table_data["ratings"].append(rating)

        # 當(dāng)天累計(jì)在線時(shí)間增加
        current_online_time += order_interval
        # 訂單序號(hào)加一
        order_cnt += 1
        # 當(dāng)天訂單數(shù)量加一
        current_order_cnt += 1

        # 當(dāng)天累計(jì)時(shí)間超過平臺(tái)限制，需要去休息
        rest_flag_1 = current_online_time >= ONLINE_DURATION_UPPER_LIMIT
        # 前面的司機(jī)追求訂單數(shù)
        rest_flag_2 = (
            driver_id < PURSUING_ORDER_DRIVER_NUM
            and current_order_cnt >= pursuing_order_volume[driver_id]
        )
        # 后面的司機(jī)追求在線時(shí)長(zhǎng)
        rest_flag_3 = (
            driver_id >= PURSUING_ORDER_DRIVER_NUM
            and current_online_time
            >= pursuing_online_duration[driver_id - PURSUING_ORDER_DRIVER_NUM]
        )

        if rest_flag_1 or rest_flag_2 or rest_flag_3:
            # 增加休息時(shí)間
            reset_interval = int(get_rest_interval_simulation(driver_id, rest_cnt))
            last_time = last_time + datetime.timedelta(seconds=reset_interval)
            # 當(dāng)天累計(jì)在線時(shí)長(zhǎng)清零
            current_online_time = 0
            # 當(dāng)天累計(jì)訂單數(shù)清零
            current_order_cnt = 0
            # 休息次數(shù)加一
            rest_cnt += 1

        # 達(dá)到項(xiàng)目總體模擬結(jié)束時(shí)間，跳出
        if last_time > END_DATETIME:
            break

3.將模擬的數(shù)據(jù)轉(zhuǎn)為 pd.DataFrame 并輸出為 csv 文件；創(chuàng)建 Hive 表，并將數(shù)據(jù) load 到表中：

df = pd.DataFrame(table_data)
df["driver_id"] = "driver_" + df["driver_id"].astype("str").str.zfill(2)
df.to_csv(
    "./dwd_uber_simulation_rating_detail.csv",
    sep=",",
    encoding="utf-8-sig",
    index=False,
    header=False,
)

from pyhive import hive

# 配置連接參數(shù)
host_ip = "127.0.0.1"
port = 10000
username = "蔣點(diǎn)數(shù)分"

with hive.Connection(host=host_ip, port=port) as conn:
    cursor = conn.cursor()

    create_table_sql = """
    CREATE TABLE IF NOT EXISTS data_exercise.dwd_uber_simulation_rating_detail (
        driver_id STRING COMMENT '司機(jī)id',
        rating_time TIMESTAMP COMMENT '評(píng)價(jià)時(shí)間',
        ratings TINYINT COMMENT '評(píng)分等級(jí)，1～5 星'
    )
    COMMENT 'Uber 乘客對(duì)司機(jī)評(píng)分表，模擬數(shù)據(jù) | 文章編號(hào) 7c98d8ef'
    ROW FORMAT DELIMITED
    FIELDS TERMINATED BY ','
    STORED AS TEXTFILE
    """

    cursor.execute(create_table_sql)

    import os

    load_data_sql = f"""
    LOAD DATA LOCAL INPATH "{os.getcwd() + '/dwd_uber_simulation_rating_detail.csv'}" 
    OVERWRITE INTO TABLE data_exercise.dwd_uber_simulation_rating_detail
    """

    cursor.execute(load_data_sql)

4.將查詢的 SQL，利用 pd.read_sql_query 讀取查詢結(jié)果。注意此段代碼，仍然位于 with 上下文中：

    select_data_sql = '''
        with calc_table as (
            select
              driver_id, date_format(rating_time, 'yyyy年ww周') as year_week -- 從周日開始算新的一周
            , sum(if(ratings <> 5, 1, 0)) over(partition by driver_id, date_format(rating_time, 'yyyy年ww周') order by rating_time asc) as cnt_tag
            , ratings
            from data_exercise.dwd_uber_simulation_rating_detail
        ) 
        
        , calc_continuous_five_table as (
            select
              driver_id, year_week, cnt_tag
            , sum(1) as continuous_five -- sum(if(raings=5,1,0))
            , rank() over(partition by year_week order by sum(1) desc) as rk
            from calc_table
            where ratings = 5
            group by driver_id, year_week, cnt_tag
        )
        
        select
          year_week
        -- 可能有司機(jī)并列，使用 collect
        -- 如果一名司機(jī)連續(xù) 5 星的次數(shù)最高，且出現(xiàn)了兩次，那么會(huì)重復(fù)
        -- 因此使用 set
        , collect_set(driver_id) as most_continuous_five_start_drivers
        from calc_continuous_five_table
        where rk = 1
        group by year_week
    '''

    df_outcome = pd.read_sql_query(select_data_sql, conn)

# 在 Jupter 環(huán)境下，顯示結(jié)果
display(df_outcome)

我通過使用 PyHive 包實(shí)現(xiàn) Python 操作 Hive。我個(gè)人電腦部署了 Hadoop 及 Hive，但是沒有開啟認(rèn)證，企業(yè)里一般常用 Kerberos 來進(jìn)行大數(shù)據(jù)集群的認(rèn)證。

四、SQL 解答

我采用 CTE 的寫法來將嵌套邏輯轉(zhuǎn)為串行，這樣寫對(duì)于復(fù)雜邏輯的結(jié)構(gòu)梳理具有一定幫助。使用窗口函數(shù) count(if(rating<>5,rating,null)) 或 sum(if(rating<>5,1,0)) 來統(tǒng)計(jì) 1~4 星評(píng)價(jià)的數(shù)量。

“每周”因此需要使用 date_format 來提取年份和周 => partition by driver_id, date_format(rating_time, 'yyyy年ww周')；使用 order by rating_time asc 時(shí)，統(tǒng)計(jì)的窗口范圍默認(rèn)是 rows between preceding unbounded and current row，寫清楚更好。

注意因?yàn)榻y(tǒng)計(jì)的邏輯是截至當(dāng)前行，所以第一個(gè) 5 星評(píng)價(jià)前的那個(gè) 1～4 星，它的計(jì)數(shù)標(biāo)識(shí)跟 5 星是一樣的。所以需要 where 過濾，當(dāng)然也可以在后續(xù)聚合統(tǒng)計(jì)時(shí)，使用條件處理 sum(if(raings=5,1,0))。

最終結(jié)果使用 collect_set 將 driver_id 形成去重?cái)?shù)組：一方面可能每個(gè)星期有司機(jī)連續(xù) 5 星好評(píng)數(shù)并列第一；另一方面極端情況下，連續(xù) 5 星好評(píng)最多的那個(gè)司機(jī)如果最多的連續(xù) 5 星好評(píng)數(shù)一周內(nèi)出現(xiàn)了多次，則這個(gè) driver_id 會(huì)出現(xiàn)多次，這是為什么不用 collect_list 的原因。

with calc_table as (
    select
      driver_id, date_format(rating_time, 'yyyy年ww周') as year_week -- 從周日開始算新的一周
    , sum(if(ratings <> 5, 1, 0)) over(partition by driver_id, date_format(rating_time, 'yyyy年ww周') order by rating_time asc) as cnt_tag
    , ratings
    from data_exercise.dwd_uber_simulation_rating_detail
) 

, calc_continuous_five_table as (
    select
      driver_id, year_week, cnt_tag
    , sum(1) as continuous_five -- sum(if(raings=5,1,0))
    , rank() over(partition by year_week order by sum(1) desc) as rk
    from calc_table
    where ratings = 5
    group by driver_id, year_week, cnt_tag
)

select
  year_week
-- 可能有司機(jī)并列，使用 collect
-- 如果一名司機(jī)連續(xù) 5 星的次數(shù)最高，且出現(xiàn)了兩次，那么會(huì)重復(fù)
-- 因此使用 set
, collect_set(driver_id) as most_continuous_five_start_drivers
, max(continuous_five) as continuous_five
from calc_continuous_five_table
where rk = 1
group by year_week

需要注意的是，date_format 的 w 參數(shù)是從周日開始算新的一周。我這里偷懶就不改成按照周一為新的一周來計(jì)算。

最簡(jiǎn)單的思路是將實(shí)際日期往前挪一天，但是周數(shù)與跨年問題，往往容易引起混淆，實(shí)際使用時(shí)需要小心處理。嚴(yán)謹(jǐn)起見，應(yīng)查詢 ISO 8601 的規(guī)定。

————

??????我現(xiàn)在正在求職數(shù)據(jù)類工作（主要是數(shù)據(jù)分析或數(shù)據(jù)科學(xué)）；如果您有合適的機(jī)會(huì)，懇請(qǐng)您與我聯(lián)系，即時(shí)到崗，不限城市。您可以發(fā)送私信或通過公眾號(hào)聯(lián)系我（全網(wǎng)同名：蔣點(diǎn)數(shù)分）。