大数据开发学习Day8

一、Linux / Shell任务批量删除当前目录下所有 .tmp 结尾的临时文件find.-name*.tmp|xargsrm-f# find . -name *.tmp 生成当前目录下所有.tmp文件的列表批量删除文件# 查找并删除所有.log文件find.-name*.log|xargsrm# 更安全的方式处理空格文件名find.-name*.log-print0|xargs-0rm批量复制文件# 复制文件到指定目录find.-name*.txt|xargs-I{}cp{}/target/dir/# 移动特定类型的文件ls*.jpg|xargs-I{}mv{}images/-I{}替换字符串xargs-I{}cp{}/dest/-nN 每次传递N个参数xargs-n1逐个处理-PN 并行处理N个进程xargs-P44进程并行-0以null分隔配合find -print0find...-print0|xargs-0-p交互模式执行前确认xargs-prmxargs把管道传来的多行文本转为命令行参数rm -f强制删除不提示大数据场景常用批量清理日志、临时表、小文件二、SQL百分比计算、累积求和、分组条件筛选1728. 每月活跃用户数 II日期 分组 去重将原始 Activity 表按 (user_id, 月份) 分组统计每个用户每月的不同活跃日期数WITHuser_daysAS(SELECTuser_id,DATE_FORMAT(activity_date,%Y-%m)ASmonth,COUNT(DISTINCTactivity_date)ASdaysFROMActivityGROUPBYuser_id,month)SELECTmonth,COUNT(DISTINCTuser_id)ASactive_usersFROMuser_daysWHEREdays2GROUPBYmonth;DATE_FORMAT(activity_date, ‘%Y-%m’)把日期如 2026-04-09格式化成 ‘2026-04’用于按月聚合1204. 最后一个能进入电梯的人累积和 上限判断WITHweight_sumAS(SELECTname,SUM(weight)OVER(ORDERBYturn)AStotalFROMQueue)SELECTnameFROMweight_sumWHEREtotal1000ORDERBYtotalDESCLIMIT1;SUM(weight) OVER (ORDER BY turn) 是窗口函数ORDER BY turn 确保按排队顺序累加550. 游戏玩法分析 IV次日留存率面试必考WITHfirst_loginAS(SELECTplayer_id,MIN(event_date)ASfirst_dateFROMActivityGROUPBYplayer_id)SELECTROUND(COUNT(DISTINCTa.player_id)/COUNT(DISTINCTf.player_id),2)ASfractionFROMfirst_login fLEFTJOINActivity aONf.player_ida.player_idANDa.event_dateDATE_ADD(f.first_date,1);CTE 找首登日期MIN(event_date) 定位每个玩家第一天登录次日判断DATE_ADD(日期, 1) 精准匹配第二天留存率公式次日登录人数 / 总玩家数去重计数必须用 COUNT(DISTINCT) 避免同一天多次登录干扰 ROUND保留 2位小数面试标准输出格式三、PySpark 核心新内容自定义函数 Pandas UDF 数据倾斜实战 分区优化在PySpark分布式计算中数据倾斜某些分区数据量过大是常见瓶颈会导致任务执行缓慢甚至失败。Pandas UDFUser Defined Function结合分区优化能高效缓解此问题它利用Spark的分布式架构将数据分块处理为pandas DataFrame在分区级别应用向量化操作减少JVM与Python的通信开销同时通过自定义分区策略确保负载均衡直接在 PyCharm 运行frompyspark.sqlimportSparkSessionfrompyspark.sqlimportfunctionsasFfrompyspark.sql.functionsimportpandas_udffrompyspark.sql.typesimportDoubleTypeimportpandasaspd sparkSparkSession.builder \.master(local[*])\.appName(day8)\.getOrCreate()# 构造数据data[(1,20.0),(2,30.0),(3,40.0),(4,50.0)]dfspark.createDataFrame(data,[id,score])# 新知识点1Pandas UDF性能远高于普通UDFpandas_udf(DoubleType())defnorm_score(s:pd.Series)-pd.Series:returns/100dfdf.withColumn(norm_score,norm_score(F.col(score)))df.show()# 新知识点2重分区 合并小文件 # 增加分区df_repartdf.repartition(4)# 合并分区df_coalescedf_repart.coalesce(1)# 新知识点3数据倾斜常用解法加盐聚合 df_saltdf.withColumn(salt,F.floor(F.rand()*3))df_salt.groupBy(id,salt).count().show()spark.stop()结果------------------|id|score|norm_score|------------------|1|20.0|0.2||2|30.0|0.3||3|40.0|0.4||4|50.0|0.5|------------------Pandas UDF 用法基于 Arrow 格式速度是普通 UDF 的 10~100 倍repartition 与 coalesce 区别repartition(n)全量重分区可增可减有 shufflecoalesce(n)只减不增无 shuffle用于合并小文件加盐 两阶段聚合解决数据倾斜随机前缀 两阶段聚合解决单分区爆量分区数对性能的影响小知识点生产环境禁止大量小文件必须用 coalesce 合并四、算法LeetCode 3. 无重复字符的最长子串掌握滑动窗口双指针写法理解用哈希集合 / 字典维护窗口时间复杂度 O (n)deflengthOfLongestSubstring(s:str)-int:char_map{}left0max_len0forright,cinenumerate(s):ifcinchar_mapandchar_map[c]left:leftchar_map[c]1char_map[c]right max_lenmax(max_len,right-left1)returnmax_len