一、Linux / Shell 综合进阶

基于之前的 access.log，完成一行命令实现：
1.筛选出状态码为 200 的行
2.提取第 3 列流量数值
3.对流量求和

cat access.log | awk '$2==200 {print $3}' | awk '{sum+=$1} END {print sum}'
awk '$2==200 {sum+=$3} END {print sum}' access.log
grep ' 200 ' access.log | awk '{sum+=$3} END {print sum}'.

小知识点
管道符 “ | ” 把前一个命令输出作为后一个输入
大数据日志排查几乎全是这类组合命令

二、SQL

1158. 市场分析 I（多表连接 + 条件统计）

SELECT 
    u.user_id AS buyer_id,
    u.join_date,
    IFNULL(COUNT(o.order_id), 0) AS orders_in_2019
FROM Users u
LEFT JOIN Orders o 
    ON u.user_id = o.buyer_id 
    AND YEAR(o.order_date) = 2019
GROUP BY u.user_id, u.join_date

1587. 花费少于既定金额的用户（分组 + HAVING 筛选）

SELECT u.name, SUM(t.amount) AS balance
FROM Users u
JOIN Transactions t ON u.account = t.account
GROUP BY u.account, u.name
HAVING SUM(t.amount) > 10000;

1285. 收入连续的时间段（区间连续问题，面试高频）

WITH tmp AS (
    SELECT *,
           id - ROW_NUMBER() OVER (PARTITION BY income_flag ORDER BY day) AS grp
    FROM (
        SELECT *,
               CASE WHEN income > 10000 THEN 1 ELSE 0 END AS income_flag
        FROM Table
    ) t
)
SELECT MIN(day) AS start_date, MAX(day) AS end_date
FROM tmp
WHERE income_flag = 1
GROUP BY grp

*今日核心新知识点：
1.分组后过滤 HAVING 与 WHERE 区别

在 SQL 查询中，WHERE 和 HAVING 都用于数据过滤，但它们在分组操作（GROUP BY）后的作用有本质区别。核心差异在于执行时机和过滤对象：WHERE 在分组前过滤单个数据行，而 HAVING 在分组后过滤整个分组。以下是详细对比（基于引用内容整理）：

• 执行时机不同

  • WHERE：在数据分组（GROUP BY）之前执行，过滤的是原始数据表中的行。它直接作用于数据库表的字段，筛选出符合条件的行，然后再进行分组

    • 例如：SELECT department, AVG(salary) FROM employees WHERE salary > 5000 GROUP BY department
    • 这里 WHERE salary > 5000 先过滤掉工资低于 5000 的行，再对剩余行分组

  • HAVING：在数据分组（GROUP BY）之后执行，过滤的是分组后的组。它针对的是分组聚合的结果

    • 例如：SELECT department, AVG(salary) FROM employees GROUP BY department HAVING AVG(salary) > 10000
    • 这里 HAVING AVG(salary) > 10000 在分组完成后，过滤出平均工资大于 10000 的部门组

  • 关键点：WHERE 影响分组前的数据输入，HAVING 影响分组后的输出结果

• 能否使用聚合函数

  • WHERE：不能使用聚合函数（如 SUM(), AVG(), COUNT()），因为它处理的是单行数据，而非分组结果
    • 错误示例：SELECT department FROM employees WHERE AVG(salary) > 10000（无效，会报错）
  • HAVING：可以使用聚合函数，因为它操作的是分组后的聚合值
    • 正确示例：SELECT department FROM employees GROUP BY department HAVING AVG(salary) > 10000（过滤平均工资高的组）
  • 原因：聚合函数依赖于分组结果，WHERE 执行时分组尚未发生

• 过滤对象不同

  • WHERE：针对数据库表的原始字段进行过滤（如 salary、age），不依赖查询结果的别名
  • HAVING：针对查询结果的输出字段（包括聚合值和别名）进行过滤
    • 示例：SELECT department AS dept, AVG(salary) AS avg_sal FROM employees GROUP BY dept HAVING avg_sal > 10000
    • 这里 HAVING 使用了别名 avg_sal，而 WHERE 无法使用别名（如 WHERE dept = ‘IT’ 会报错）

• 在分组查询中的典型流程
  SQL 查询的执行顺序为：WHERE → GROUP BY → HAVING → SELECT（输出结果）
  分组前过滤：用 WHERE 减少数据量，提高效率
  分组后过滤：用 HAVING 筛选符合条件的组

-- 查询部门平均工资超过 10000 的部门，且仅考虑全职员工（分组后过滤用 HAVING）
SELECT department, AVG(salary) AS avg_salary
FROM employees
WHERE employment_type = 'full-time'  -- 分组前过滤：WHERE 作用于单行
GROUP BY department
HAVING AVG(salary) > 10000;          -- 分组后过滤：HAVING 作用于分组

说明：WHERE 先过滤出全职员工行，然后 GROUP BY 分组，最后 HAVING 过滤出平均工资达标的组

• 总结
  • 分组后过滤必须用 HAVING：因为它直接处理分组结果，支持聚合函数和别名
  • 错误使用后果：在 HAVING 中放行级条件（如 HAVING salary > 5000）会导致逻辑错误或性能下降；在 WHERE 中用聚合函数会报错
  • 最佳实践：
    • 行级过滤用 WHERE（例如：WHERE age > 30）
    • 组级过滤用 HAVING（例如：HAVING COUNT(*) > 5）

通过合理组合 WHERE 和 HAVING，可以高效实现复杂查询。例如，电商分析中先用 WHERE 过滤特定年份订单，再用 HAVING 筛选总销售额超标的商品组

2.连续区间类问题通用解法

连续区间类问题在SQL中通常涉及识别数值序列中的连续段（如连续订单号、连续登录日期等）

• 差值分组法（最常用）
  原理：利用序列值与行号的差值标识连续区间（连续数据差值恒定）
  适用场景：查找连续区间起止点、计算连续天数

SELECT 
  MIN(log_id) AS start_id,
  MAX(log_id) AS end_id
FROM (
  SELECT 
    log_id,
    log_id - ROW_NUMBER() OVER (ORDER BY log_id) AS grp  -- 关键差值计算
  FROM Logs
) t
GROUP BY grp
ORDER BY start_id;

核心逻辑：
ROW_NUMBER() OVER (ORDER BY log_id) 生成有序行号
log_id - row_number 在连续区间内差值相同
按差值分组后取最小/最大值即为区间端点

• 边界特征法（高效检测端点）
  原理：连续区间的边界值满足 x-1不存在（起点） 或 x+1不存在（终点）

-- 查找所有连续区间端点
WITH Starts AS (
  SELECT log_id AS start_id
  FROM Logs
  WHERE NOT EXISTS (SELECT 1 FROM Logs WHERE log_id = t.log_id - 1)
),
Ends AS (
  SELECT log_id AS end_id
  FROM Logs
  WHERE NOT EXISTS (SELECT 1 FROM Logs WHERE log_id = t.log_id + 1)
)
SELECT start_id, MIN(end_id) AS end_id
FROM Starts
JOIN Ends ON start_id <= end_id
GROUP BY start_id;

• 参数化连续间隔检测
  原理：通过变量 n 自定义连续间隔（如间隔≤2天算连续）

SELECT 
  user_id,
  MIN(login_date) AS start_date,
  MAX(login_date) AS end_date
FROM (
  SELECT 
    user_id,
    login_date,
    SUM(gap_flag) OVER (PARTITION BY user_id ORDER BY login_date) AS grp
  FROM (
    SELECT 
      user_id,
      login_date,
      -- 间隔>n时标记为新组 (n=2)
      CASE WHEN DATEDIFF(login_date, LAG(login_date) OVER (PARTITION BY user_id ORDER BY login_date)) > 2 
           THEN 1 ELSE 0 END AS gap_flag
    FROM usr_login
  ) t
) t2
GROUP BY user_id, grp;

• 窗口函数检测连续出现
  适用场景：判断连续出现N次的记录（如连续3天登录）

SELECT DISTINCT Num AS ConsecutiveNums
FROM (
  SELECT 
    Num,
    LAG(Num, 1) OVER (ORDER BY Id) AS prev1,
    LAG(Num, 2) OVER (ORDER BY Id) AS prev2
  FROM Logs
) t
WHERE Num = prev1 AND Num = prev2;

• 方法对比
  

关键技巧：
所有方法本质都是 标识连续分组 → 聚合计算
大数据量时优先选差值分组法（效率最高）
含日期数据用DATEDIFF()/LAG() 处理时间连续性

3.多表左连接 + 计数为空的处理

在SQL多表左连接（LEFT JOIN）中处理计数为空的情况时，核心问题在于保留左表所有记录的同时，正确处理右表无匹配时的NULL值。以下是系统化的解决方案：

• 基础处理：COALESCE函数转换NULL

适用场景：简单计数场景，右表无匹配时显示0而非NULL

SELECT 
  t1.id,
  COALESCE(COUNT(t2.foreign_key), 0) AS valid_count  -- NULL转为0
FROM table1 t1
LEFT JOIN table2 t2 ON t1.id = t2.foreign_key
GROUP BY t1.id;

原理：COALESCE(expression, 0) 将NULL替换为0

• 子查询预聚合：避免连接膨胀

适用场景：多表连接导致计数翻倍

SELECT 
  t1.id,
  COALESCE(pre_agg.sub_count, 0) AS safe_count  -- 子查询预聚合
FROM table1 t1
LEFT JOIN (
  SELECT foreign_key, COUNT(*) AS sub_count
  FROM table2 
  GROUP BY foreign_key  -- 先聚合再连接
) pre_agg ON t1.id = pre_agg.foreign_key;

优势：
避免连接多对多关系导致的计数重复
提升大表查询性能

• 多表连接计数：SUM+CASE组合

适用场景：多个右表需要独立计数

SELECT
  t1.id,
  SUM(CASE WHEN t2.id IS NOT NULL THEN 1 ELSE 0 END) AS count_t2,
  SUM(CASE WHEN t3.id IS NOT NULL THEN 1 ELSE 0 END) AS count_t3
FROM table1 t1
LEFT JOIN table2 t2 ON t1.id = t2.fk
LEFT JOIN table3 t3 ON t1.id = t3.fk
GROUP BY t1.id;

关键点：
每个右表使用独立CASE判断存在性
SUM()聚合替代COUNT()直接计数

• 全外连接处理：FULL OUTER JOIN

适用场景：需要包含所有表的完整记录

SELECT 
  COALESCE(t1.id, t2.fk) AS id,
  COUNT(t2.fk) AS count_t2  -- 自动处理NULL
FROM table1 t1
FULL OUTER JOIN table2 t2 ON t1.id = t2.fk
GROUP BY COALESCE(t1.id, t2.fk);

注意：MySQL需改用UNION模拟全外连接

• 方案对比表
  

避坑指南：
始终优先考虑子查询预聚合，避免连接膨胀问题
明确业务需求：是否需要统计NULL？用COUNT(column)忽略NULL，COUNT()包含NULL
多表连接时使用SELECT DISTINCT去重可能导致计数错误*

三、PySpark 核心进阶

承接 Day5 UDF、加盐，今日新内容：持久化级别、数据倾斜处理、函数高阶用法

直接在 PyCharm 运行并理解：

from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.storagelevel import StorageLevel

spark = SparkSession.builder \
    .master("local[*]") \
    .appName("day6") \
    .getOrCreate()

# 测试数据
data = [
    (1, "click", "2025-01-01"),
    (1, "click", "2025-01-01"),
    (2, "view", "2025-01-01"),
    (3, "buy", "2025-01-02")
]
df = spark.createDataFrame(data, ["uid", "event", "dt"])

# ==================== 新知识点1：持久化级别 ====================
# 对比cache，指定存储级别
df.persist(StorageLevel.MEMORY_AND_DISK)
# 释放缓存
# df.unpersist()

# ==================== 新知识点2：时间函数 ====================
df.withColumn("year", F.year(F.col("dt"))) \
  .withColumn("month", F.month(F.col("dt"))) \
  .show()

# ==================== 新知识点3：分组统计+去重 ====================
df.groupBy("uid", "event").count().show()
df.dropDuplicates(["uid", "event", "dt"]).show()

# ==================== 新知识点4：过滤条件多条件 ====================
df.filter((F.col("uid") == 1) & (F.col("event") == "click")).show()

# ==================== 新知识点5：数据倾斜处理-打散key ====================
# 加盐
df = df.withColumn("salt", F.concat(F.col("uid"), F.lit("_"), F.floor(F.rand() * 2)))
df.show()

spark.stop()

今日必须掌握：
1.persist 与 cache 区别

在PySpark中，persist()和cache()都是用于优化RDD/DataFrame计算性能的数据持久化方法，但存在关键差异：

• 存储级别灵活性
  
  存储级别选项：

from pyspark import StorageLevel

# 常用存储级别
StorageLevel.MEMORY_ONLY         # 仅内存
StorageLevel.MEMORY_AND_DISK     # 内存+磁盘溢出
StorageLevel.MEMORY_ONLY_SER     # 内存（序列化）
StorageLevel.DISK_ONLY           # 仅磁盘

• 方法本质

# cache() 本质是 persist() 的快捷方式
def cache(): 
    return persist(StorageLevel.MEMORY_ONLY)  # [^4]
    

• 使用场景对比
  
• 代码示例

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("persist-demo").getOrCreate()
df = spark.read.csv("large_dataset.csv")

# cache() 用法
df.cache()  # 默认存储到内存

# persist() 用法
df.persist(StorageLevel.MEMORY_AND_DISK)  # 内存不足时存磁盘

• 性能注意事项

  • 惰性执行：两者都是惰性操作，遇到Action操作（如count(), collect()）才会真正持久化
    df.cache() # 此时无实际操作
    df.count() # 触发缓存执行
  • 内存管理：
    • cache()在内存不足时会被淘汰，导致重算
    • persist()可通过磁盘备份避免此问题
  • 释放缓存：
    df.unpersist() # 两者通用释放方法

• 最佳实践建议

  • 优先使用persist()：更灵活应对数据规模和集群资源
  • 监控存储级别：
    • df.storageLevel # 查看当前存储级别
    • #输出：StorageLevel(True, True, False, False, 1)
  • 避免过度缓存：仅缓存会被重复使用的中间结果

2.时间函数处理

在PySpark中，时间函数主要通过pyspark.sql.functions模块提供，支持丰富的日期/时间处理能力。以下是核心处理方式：

• 基础时间函数

from pyspark.sql import functions as F

# 创建示例数据
df = spark.createDataFrame([("2023-08-01 14:30:00",)], ["timestamp"])

# 常用基础函数
df.select(
    F.current_date(),                  # 当前日期 → 2023-08-15
    F.current_timestamp(),             # 当前时间戳 → 2023-08-15T08:45:22.123Z
    F.year("timestamp"),               # 提取年份 → 2023
    F.month("timestamp"),              # 提取月份 → 8
    F.dayofmonth("timestamp"),         # 提取日期 → 1
    F.hour("timestamp"),               # 提取小时 → 14
    F.date_format("timestamp", "MM/dd/yyyy")  # 格式化 → 08/01/2023
).show()

• 时间转换与解析
  

# 时区转换
df.select(
    F.from_utc_timestamp("timestamp", "Asia/Shanghai")  # UTC转上海时间
)

• 时间计算

# 日期计算
df.select(
    F.date_add("timestamp", 7),        # 加7天 → 2023-08-08
    F.date_sub("timestamp", 3),        # 减3天 → 2023-07-29
    F.datediff(F.current_date(), "timestamp")  # 日期差 → 14天
)

# 时间差计算
df.withColumn("duration", 
    F.unix_timestamp("end_time") - F.unix_timestamp("start_time")  # 秒级差值
)

• 窗口时间处理

from pyspark.sql.window import Window

# 按时间窗口聚合
window_spec = Window.orderBy("timestamp").rangeBetween(-86400, 0)  # 24小时窗口

df.withColumn("24h_avg", 
    F.avg("value").over(window_spec)
)

• 时间序列特征工程

# 提取时间特征
df = df.withColumn("day_of_week", F.dayofweek("timestamp"))  # 周几(1=周日)
        .withColumn("is_weekend", F.when(F.dayofweek("timestamp") > 5, 1).otherwise(0))
        .withColumn("hour_sin", F.sin(2 * F.pi * F.hour("timestamp")/24))  # 周期性编码

• 处理常见问题

格式错误处理

df.withColumn("safe_date", 
    F.to_date(F.col("date_str"), "yyyy-MM-dd")  # 格式不匹配返回null
)

时区统一

spark.conf.set("spark.sql.session.timeZone", "UTC")  # 设置会话时区

性能优化

# 避免在UDF中使用时间函数，优先使用内置函数[^3]
# 错误方式
@F.udf
def extract_year_udf(ts):
    return ts.year  # 低效

# 正确方式
df.select(F.year("timestamp"))  # 内置函数优化

3.多条件过滤写法

在PySpark中，多条件过滤主要通过filter()或where()函数（两者等价）实现，结合逻辑运算符组合多个条件。以下是5种常用写法：

• 基础逻辑运算符组合（最常用）

from pyspark.sql import functions as F

# AND条件（与）
df_filtered = df.filter(
    (F.col("age") > 30) & 
    (F.col("department") == "IT")
)

# OR条件（或）
df_filtered = df.filter(
    (F.col("salary") < 5000) | 
    (F.col("job_level") == "Junior")
)

# NOT条件（非）
df_filtered = df.filter(
    ~(F.col("status").isin("离职", "退休"))  # 排除离职和退休人员
)

注意：每个条件必须用括号包裹，避免运算符优先级问题

• 链式调用（分步过滤）

df_filtered = df.filter(F.col("sales") > 1000) \
               .filter(F.col("region") == "East") \
               .filter(F.year("order_date") == 2023)

适用于需要逐步缩小数据范围的场景

• SQL表达式字符串（简化语法）

df_filtered = df.filter(
    "age >= 25 AND (department = 'HR' OR department = 'Finance')"
)

优点：语法简洁，类似SQL的WHERE子句

• when()函数组合（复杂条件分支）

from pyspark.sql.functions import when

df_filtered = df.filter(
    when(F.col("tenure") > 5, F.col("salary") > 8000)
    .otherwise(F.col("salary") > 5000)
)

适用于需要条件分支的过滤逻辑

• 动态条件构建（编程式生成）

conditions = [
    F.col("product_type") == "Electronics",
    F.col("rating") >= 4.0
]

# 动态组合AND条件
final_condition = conditions[0]
for cond in conditions[1:]:
    final_condition = final_condition & cond

df_filtered = df.filter(final_condition)

适用于条件数量动态变化的场景

4.打散 key 缓解数据倾斜

在PySpark中，打散Key（Salting）是解决数据倾斜的核心技术，通过为Key添加随机前缀/后缀，将大Key拆分成多个小Key分散处理。以下是三种常用方法：

• 分离倾斜Key单独处理（针对少数大Key）

适用场景：少数Key导致倾斜（如个别热点用户）

from pyspark.sql import functions as F

# 步骤1：识别并分离热点Key（如key='VIP_USER'）
hot_key = "VIP_USER"
df_skew = df.filter(F.col("key") == hot_key)  # 热点数据
df_normal = df.filter(F.col("key") != hot_key)  # 正常数据

# 步骤2：为热点Key添加随机前缀（0-9）
df_skew_salted = df_skew.withColumn("salted_key", 
    F.concat(F.lit("salt_"), F.floor(F.rand(seed=42)*10), F.lit("_"), F.col("key"))
)

# 步骤3：构建扩容的维度表（10倍扩容）
dimension_expanded = dimension_df.crossJoin(
    spark.range(0, 10).toDF("salt_prefix")
).withColumn("salted_key", 
    F.concat(F.col("salt_prefix"), F.lit("_"), F.col("key"))
)

# 步骤4：分别Join后合并
result_skew = df_skew_salted.join(dimension_expanded, "salted_key")
result_normal = df_normal.join(dimension_df, "key")
final_result = result_normal.unionByName(result_skew.drop("salted_key"))

• 整体随机打散（针对多个倾斜Key）

适用场景：倾斜Key较多，无法逐个处理

# 步骤1：为事实表所有Key添加随机前缀
salt_num = 100  # 打散倍数
df_salted = df.withColumn("salt", F.floor(F.rand() * salt_num)) \
              .withColumn("salted_key", F.concat(F.col("salt"), F.lit("_"), F.col("key")))

# 步骤2：维度表扩容salt_num倍
dim_salted = dimension_df.withColumn("join_flag", F.lit(1)) \
    .crossJoin(spark.range(0, salt_num).toDF("salt")) \
    .withColumn("salted_key", F.concat(F.col("salt"), F.lit("_"), F.col("key")))

# 步骤3：基于新Key进行Join
result = df_salted.join(dim_salted, "salted_key") \
                 .drop("salt", "salted_key", "join_flag")

• 聚合操作双重打散（Group By优化）

适用场景：聚合操作数据倾斜（类似Hive的skewindata）

# 阶段1：局部聚合（添加随机盐）
salt_num = 10
df_stage1 = df.withColumn("salt", F.floor(F.rand() * salt_num)) \
              .groupBy("key", "salt") \
              .agg(F.sum("value").alias("partial_sum"))

# 阶段2：全局聚合（去除盐值）
df_stage2 = df_stage1.groupBy("key") \
                     .agg(F.sum("partial_sum").alias("total_sum"))

• 方法选型指南
  

小知识点
cache( ) = persist (StorageLevel.MEMORY_ONLY)，内存不足时会丢失数据

四、算法

今日递进：哈希表 + 链表经典题

LeetCode 160. 相交链表
要求：
1.掌握双指针最优解法
2.理解空间 O (1) 思路
3.对比哈希表暴力解法

思路要点：
1.两个指针分别遍历两条链表，走到末尾后交换起点
2.相遇点即为相交节点

class ListNode:
    def __init__(self, x):
        self.val = x
        self.next = None

class Solution:
    def getIntersectionNode(self, headA: ListNode, headB: ListNode) -> ListNode:
        p1, p2 = headA, headB
        while p1 != p2:
            p1 = p1.next if p1 else headB
            p2 = p2.next if p2 else headA
        return p1