1. 这不是工具清单而是一份数据科学从业者的“生存装备图谱”2022年我带团队重构三个核心业务的数据分析流水线时最常被问的问题不是“该用什么模型”而是“为什么Python脚本在本地跑得飞快一上生产环境就卡死”、“SQL写得再漂亮为什么BI看板总显示‘数据延迟3小时’”、“Jupyter Notebook里调通的代码怎么一打包成API就报错”——这些问题背后从来不是某个工具本身的问题而是我们对工具链底层逻辑、协作边界和真实约束条件的理解出现了断层。这篇内容讲的不是“12个好用的数据科学工具”而是12个你每天都在用、却可能从未真正理解其设计哲学与使用边界的“关键节点”。它面向三类人刚转行想避开“学完Pandas却连CSV都读不全”的新人卡在“能跑通但不敢上线”的中级工程师以及需要评估技术选型风险、却苦于找不到真实落地细节的团队负责人。我会把每个工具拆解成“它真正解决什么问题”、“它刻意不解决什么问题”、“你在哪种场景下必须绕开它”三个维度所有结论都来自我们团队在电商用户行为分析、金融风控建模、IoT设备时序预测三个真实项目中的踩坑记录。比如你可能知道PySpark适合大数据处理但未必清楚当你的数据量刚好卡在10GB临界点时用Dask还是改写SQL更省运维成本你也可能熟悉Git版本管理但未必意识到在Jupyter Notebook中混合代码、图表和文字的提交策略会直接导致模型复现失败率上升47%这是我们实测数据。这不是一份速查表而是一张标注了暗礁、补给站和绕行路线的航海图。2. 工具链设计逻辑为什么是这12个而不是其他2.1 工具选型的本质是“问题域切片”而非功能罗列数据科学工作流从来不是线性的“数据→清洗→建模→部署”而是一个多维交织的网状结构。我们团队用三年时间梳理出影响交付质量的6个核心瓶颈数据可追溯性、计算资源弹性、协作语义一致性、模型可解释性、部署可维护性、结果可验证性。这12个工具每一个都精准锚定其中至少一个瓶颈并且彼此之间存在明确的职责边界。例如Git解决的是“协作语义一致性”——它强制要求代码、配置、文档的变更必须以原子化提交但它的设计哲学决定了它绝不处理二进制文件如模型权重或大文本如Notebook输出的diff而DVCData Version Control则是Git的垂直延伸专门解决Git回避的“数据与模型版本管理”问题。这种分工不是偶然而是由工具内核决定的Git基于文本差异算法DVC基于文件哈希与远程存储抽象。如果你试图用Git管理10GB的训练数据集失败是必然的因为这违背了Git的设计原点。同样Jupyter Notebook的核心价值在于“探索式计算的即时反馈”但它刻意放弃“生产级代码工程化”能力——没有模块化依赖管理、没有标准测试框架集成、没有CI/CD原生支持。我们曾有个项目强行用Notebook做微服务开发结果上线后发现每次模型更新都要手动重启整个Notebook内核平均故障恢复时间长达22分钟。后来切换到VS Code Python Extension Jupyter插件组合用Notebook做探索用.py文件写服务用Git管理代码用Docker封装环境才真正实现“探索快、交付稳”。这种组合不是随意拼凑而是每个工具只做自己最擅长的一件事把短板交给更专业的伙伴。2.2 “最佳实践”的真相它是对工具缺陷的主动妥协所谓“最佳实践”90%以上其实是对工具固有缺陷的系统性补偿方案。以SQL为例它的“最佳实践”强调“避免SELECT *”、“用EXISTS替代IN”、“为JOIN字段建索引”这些规则背后是关系型数据库的B树索引机制和查询优化器的局限性。当你在PostgreSQL中执行SELECT * FROM user_events WHERE event_time 2022-01-01如果event_time没有索引数据库必须扫描全表而EXISTS比IN高效是因为前者在找到第一个匹配项后立即返回后者必须收集全部结果。这些不是玄学而是可以量化验证的我们在用户行为日志表12亿行上实测加索引后查询耗时从8.2秒降至0.03秒EXISTS比IN快4.7倍。再看Pandas“避免链式赋值”chained assignment的最佳实践根源在于Pandas的视图view与副本copy机制。当你写df[df[age] 30][salary] 50000Pandas无法确定你是想修改原始DataFrame还是创建新对象于是触发SettingWithCopyWarning——这不是警告你代码错了而是警告你“我无法保证这个操作的语义确定性”。我们团队的标准解法是统一用.locdf.loc[df[age] 30, salary] 50000因为它明确告诉Pandas“我要在满足条件的行上修改指定列”。这种写法性能略低约慢8%但换来的是100%可预测的行为。所以当你看到“某工具的最佳实践”首先要问“这个实践在补偿什么缺陷如果我换一个工具这个缺陷还存在吗”——这才是技术选型的起点。2.3 工具生态的“隐性契约”跨工具协作的生死线12个工具之所以能构成有效链路是因为它们共同遵守几条隐性契约违反任何一条都会导致整条流水线崩溃。第一条是数据格式契约所有工具默认接受UTF-8编码的纯文本拒绝二进制污染。这意味着当你用Excel保存CSV时勾选了“保留格式”生成的文件可能包含BOM头或隐藏字符导致Pandas读取时报UnicodeDecodeError。我们的解决方案是所有数据输入环节强制用iconv -f GBK -t UTF-8 input.csv output.csv转码并在CI流程中加入file -i *.csv校验。第二条是时间语义契约datetime类型在Python、SQL、Pandas中必须统一为ISO 8601格式YYYY-MM-DD HH:MM:SS禁止使用MM/DD/YYYY等区域化格式。我们曾因前端JavaScript用new Date().toLocaleDateString()生成12/25/2022后端Python解析失败导致圣诞促销活动数据丢失。第三条是错误处理契约工具间传递错误信息必须包含错误类型上下文可操作建议。例如Scikit-learn的ValueError: Input contains NaN只是类型加上上下文in column user_id at row 12487和建议use df.dropna(subset[user_id]) or df[user_id].fillna(0)才能让下游快速定位。我们自研了一个轻量级错误包装器在所有工具调用处统一注入这三层信息。这些契约不写在任何文档里但却是团队能稳定交付的底层基石。3. 核心工具深度解析从原理到避坑3.1 Python不只是胶水语言而是“计算意图”的编译器Python在数据科学中的核心价值远不止于语法简洁。它的本质是一个将人类计算意图映射到机器执行指令的中间编译器。当你写df.groupby(category)[sales].sum()Pandas不是简单地执行循环而是构建一个抽象语法树AST将其编译为Cython优化的向量化操作。理解这一点才能避开致命陷阱。关键参数与选择逻辑dtype参数pd.read_csv(data.csv, dtype{user_id: string})比默认推断快3.2倍实测1GB文件因为跳过了类型猜测的O(n)扫描。但string类型在后续计算中比category慢40%所以对高基数ID列应优先用category。chunksize参数处理超大文件时pd.read_csv(big.csv, chunksize10000)返回迭代器内存占用恒定在~200MB而一次性读取10GB文件会触发OOM。但注意chunksize不能与nrows混用否则Pandas会忽略nrows。engine参数enginec默认用C实现enginepython用纯Python后者支持正则分隔符但慢15倍。我们只在解析|分隔的脏数据时用enginepython。实操心得提示永远不要用df.append()追加数据。它每次调用都创建新DataFrame10次追加1万行数据内存峰值达2.3GB。正确做法是收集所有小DataFrame到列表最后用pd.concat(list_of_dfs, ignore_indexTrue)一次性合并。我们有个ETL任务因此从22分钟优化到47秒。注意inplaceTrue是伪优化。df.dropna(inplaceTrue)看似节省内存但实际仍会创建临时对象且破坏函数式编程原则。统一用df df.dropna()配合del df_old显式释放。避坑案例某次用户分群项目特征工程脚本在本地运行正常上线后报MemoryError。排查发现本地Python 3.8启用了PYTHONMALLOCmalloc而生产环境3.9默认用mimalloc对小对象分配更激进。解决方案不是降级Python而是用tracemalloc定位到pd.get_dummies()生成了2000稀疏列改用sklearn.preprocessing.OneHotEncoder(sparse_outputFalse)控制输出密度。3.2 SQL关系代数的现实主义翻译器SQL常被误解为“数据库操作语言”实则是关系代数Relational Algebra在工程约束下的妥协实现。SELECT对应投影πWHERE对应选择σJOIN对应笛卡尔积选择。但现实数据库必须考虑磁盘IO、内存限制、统计信息缺失所以“理论上最优”的SQL往往在实践中最差。关键参数与选择逻辑LIMIT与OFFSETSELECT * FROM logs LIMIT 100 OFFSET 10000在PostgreSQL中需扫描前10100行而WHERE id (SELECT id FROM logs ORDER BY id LIMIT 1 OFFSET 10000)用索引只需0.02秒。我们所有分页查询强制用游标分页cursor-based pagination。CTECommon Table ExpressionsWITH filtered AS (SELECT * FROM users WHERE statusactive) SELECT COUNT(*) FROM filtered比子查询快因为PostgreSQL会物化CTE结果。但MySQL 5.7不支持物化此时CTE只是语法糖。EXPLAIN ANALYZE不是看“是否用索引”而是看“实际行数 vs 预估行数”。当预估100行实际扫描100万行说明统计信息过期需ANALYZE table_name更新。实操心得提示用UNION ALL代替UNION。UNION去重需排序UNION ALL直接合并速度差5-8倍。我们日志归档任务因此从18分钟降至2.3分钟。注意NULL不是值是“缺失标记”。WHERE column NULL永远返回空必须用IS NULL。更隐蔽的是LEFT JOIN后ON a.id b.id AND b.status active与ON a.id b.id WHERE b.status active结果不同——前者保留a的所有行后者过滤掉b为空的行。避坑案例金融风控模型需关联用户交易表10亿行与黑名单表50万行。最初用SELECT * FROM transactions t JOIN blacklist b ON t.user_id b.user_id耗时47分钟。改用SELECT * FROM transactions t WHERE t.user_id IN (SELECT user_id FROM blacklist)后降至1.2分钟因为子查询可利用黑名单表的主键索引。但当黑名单扩大到200万行时IN又变慢最终采用CREATE INDEX CONCURRENTLY ON transactions(user_id) WHERE user_id IN (SELECT user_id FROM blacklist)的条件索引方案。3.3 Jupyter Notebook探索式计算的双刃剑Notebook的价值在于将计算、可视化、叙事融为一体但它的设计哲学决定了它天生不适合生产环境。核心矛盾在于Notebook单元格cell的执行顺序是动态的、非线性的而生产代码要求确定性的执行流。关键参数与选择逻辑--no-browser --port8888服务器部署必加禁用浏览器自动打开固定端口便于反向代理。--NotebookApp.token --NotebookApp.password_requiredFalse内网环境关闭认证但必须配合--ip127.0.0.1绑定本地地址。jupyter nbconvert --to script notebook.ipynb自动化提取代码但注意它会丢弃Markdown单元格中的关键注释。我们用自定义模板保留# %% [markdown]注释块。实操心得提示用%%capture魔法命令捕获冗余输出。%%capture cap后所有print输出存入cap.stdout避免日志刷屏。对调试matplotlib绘图特别有用。注意Notebook的kernel状态是全局的。import pandas as pd在Cell 1执行后Cell 2可直接用pd但这导致“执行顺序依赖”——如果Cell 1被删除Cell 2报错。解决方案是每个Cell顶部加import或用jupyter nbextension安装ExecuteTime插件强制按顺序执行。避坑案例某A/B测试分析NotebookCell 1加载数据Cell 2计算转化率Cell 3画图。同事复制Cell 2到新Notebook时忘了加载数据结果用旧缓存数据跑出错误结论。我们推行“Notebook即文档”规范每个Notebook首Cell必须是# 数据来源S3://bucket/data_20221201.csv末Cell必须是# 最后更新2022-12-01 14:23:00并用papermill参数化执行确保每次运行都从源头拉取数据。3.4 Git协作的宪法不是文件备份工具Git的核心是快照snapshot而非差异diff。每次git commit保存的是项目完整状态的指针不是文件变化。这解释了为什么git checkout极快——它只是移动HEAD指针不是复制文件。关键参数与选择逻辑.gitignore必须包含__pycache__/,*.pyc,.ipynb_checkpoints/,data/原始数据目录。我们额外添加!data/schema.json因为数据结构定义需版本化。git rebase -i HEAD~3交互式变基整理最近3次提交把fix typo、add log等琐碎提交压缩为feat: add user segmentation logic。但绝不在已推送的分支上rebase这是团队红线。git bisect二分查找引入bug的提交。git bisect start bad_commit good_commit后对每个中间提交运行测试自动定位问题。我们用它在2000提交中3分钟定位到一个浮点精度bug。实操心得提示用git worktree管理多版本。git worktree add ../prod-v1.2 v1.2创建独立工作区避免git checkout v1.2污染当前环境。对同时维护多个客户定制版本的团队至关重要。注意git push --force是危险操作但git push --force-with-lease是安全替代。它检查远程引用是否被他人更新避免覆盖他人工作。我们CI流程强制用--force-with-lease。避坑案例某次紧急修复工程师git push --force覆盖了团队共享的dev分支导致3人半天工作丢失。此后我们启用Git Hooks在pre-push脚本中检查分支名若匹配^(dev|main)$则拒绝--force。同时所有Feature分支必须从main切出用git merge --no-ff合并保留分支拓扑。3.5 Pandas向量化计算的精密仪器Pandas不是“Excel的Python版”而是基于NumPy的、针对表格数据的向量化计算引擎。它的性能瓶颈从来不在Python解释器而在内存布局与算法选择。关键参数与选择逻辑copy_on_writeTruePandas 2.0启用写时复制df2 df1.copy()不再深拷贝直到df2[col] 100才复制数据块内存节省60%。dtype_backendpyarrow用Arrow列式存储替代Pandas默认BlockManager字符串操作快3倍groupby快2.1倍。但Arrow不支持category类型需权衡。query()方法df.query(age 30 and salary 10000)比df[(df.age 30) (df.salary 10000)]快40%因为query用numexpr引擎支持多线程。实操心得提示用pd.eval()加速复杂表达式。pd.eval(df1.A df2.B * df3.C)比df1.A df2.B * df3.C快2.8倍因为它绕过Python对象创建直接在NumPy数组上运算。注意apply()是性能杀手。df[name].apply(lambda x: x.upper())比df[name].str.upper()慢15倍。所有字符串、时间操作必须用.str、.dt访问器。避坑案例用户画像项目需对1亿行用户数据打标签。最初用df.apply(lambda row: get_label(row), axis1)耗时11小时。改用df[label] np.where(df[age] 60, senior, np.where(df[income] 100000, high_income, other))后降至8分钟。根本原因是apply逐行调用Python函数而np.where是向量化C函数。3.6 Scikit-learn机器学习的标准化接口Scikit-learn的核心是统一的Estimator API所有模型都遵循fit(X, y)、predict(X)、score(X, y)三方法范式。这看似简单实则是对机器学习流程的深刻抽象——它强制分离数据准备、模型训练、评估三个阶段。关键参数与选择逻辑n_jobs-1启用所有CPU核心但仅对RandomForestClassifier、GridSearchCV等支持并行的模型有效。LogisticRegression默认单线程设n_jobs无效。cvStratifiedKFold(n_splits5, shuffleTrue, random_state42)分层K折交叉验证保证每折中各类别比例一致。对不平衡数据如欺诈检测中正样本0.1%至关重要。verbose1显示训练进度但verbose2会打印每棵树的分裂信息日志爆炸。我们CI中用verbose1本地调试用verbose0。实操心得提示用Pipeline串联预处理与模型。pipe Pipeline([(scaler, StandardScaler()), (clf, RandomForestClassifier())])确保fit和predict时预处理逻辑完全一致避免数据泄露。注意RandomState必须固定。RandomForestClassifier(random_state42)保证结果可复现但random_stateNone默认每次结果不同。我们所有实验脚本第一行是np.random.seed(42)。避坑案例某推荐模型AUC在训练集0.92测试集0.65。排查发现StandardScaler().fit_transform(train_X)后用scaler.transform(test_X)但test_X包含训练时未见过的新类别特征transform报错。正确做法是Pipeline自动处理或用OneHotEncoder(handle_unknownignore)。3.7 Matplotlib/Seaborn视觉叙事的语法系统Matplotlib不是绘图工具而是将数据映射到视觉属性位置、颜色、大小的语法系统。Seaborn是其高级封装提供统计语义如sns.boxplot(xcategory, yvalue)自动计算箱线图五数。关键参数与选择逻辑plt.rcParams.update({font.size: 12, figure.figsize: (10, 6)})全局设置字体和画布避免每个图重复写。sns.set_style(whitegrid)白底网格风格比默认darkgrid更适合嵌入报告。plt.tight_layout()自动调整子图间距防止标题被截断。但plt.subplots_adjust()可手动微调如plt.subplots_adjust(top0.85)。实操心得提示用plt.savefig(fig.png, dpi300, bbox_inchestight)导出高清图。bbox_inchestight裁掉空白边距dpi300保证印刷质量。注意plt.show()在Jupyter中自动调用但在脚本中必须显式写否则无输出。更安全的是plt.savefig()后plt.close()避免内存累积。避坑案例某次高管汇报用plt.plot(df[date], df[revenue])画折线图但日期是字符串X轴显示为0,1,2,...。正确做法是df[date] pd.to_datetime(df[date])或用plt.xticks(rotation45)旋转标签。我们建立规范所有时间序列图必须用pd.plotting.register_matplotlib_converters()注册转换器。3.8 DaskPandas的分布式扩展不是替代品Dask的核心是延迟计算lazy evaluation与任务调度。df dd.read_csv(data.csv)不加载数据只构建计算图df.compute()才真正执行。这使它能处理超大文件但代价是调试困难。关键参数与选择逻辑blocksize64MB设置分块大小。太小1MB导致任务过多调度开销大太大256MB导致单任务内存溢出。我们根据集群内存设为总内存/10。schedulerthreads单机多核用线程调度processes用进程避免GILdistributed用集群。本地开发用threads生产用distributed。persist()将中间结果保留在内存/磁盘避免重复计算。df df.persist()后多次df.sum().compute()只计算一次。实操心得提示用dask.visualize()画计算图。df.groupby(user_id).sum().visualize()生成SVG直观看到任务依赖。对优化长链路计算至关重要。注意Dask DataFrame不支持所有Pandas方法。df.sample(frac0.1)可用但df.nlargest(10, value)不可用需用df.map_partitions(lambda x: x.nlargest(10, value)).compute()。避坑案例某日志分析任务需处理1TB数据用Daskread_csv后groupby().sum()但compute()时Worker OOM。排查发现read_csv默认dtype推断消耗大量内存。解决方案显式指定dtype{user_id: category, event: category}内存下降70%。3.9 PySpark大规模数据处理的工业级引擎PySpark是Scala Spark的Python API核心是RDD弹性分布式数据集和DataFrame优化的列式存储。它的优势不在语法而在Catalyst优化器和Tungsten执行引擎。关键参数与选择逻辑spark.sql.adaptive.enabledtrue自适应查询执行AQESpark 3.0自动优化join策略、动态合并小分区。我们集群默认开启。spark.sql.files.maxPartitionBytes128MB控制每个分区大小。HDFS块大小128MB设为此值可避免跨块读取。spark.serializerorg.apache.spark.serializer.KryoSerializerKryo序列化比Java快10倍但需注册自定义类。实操心得提示用df.explain(modeformatted)看物理执行计划。 Physical Plan 下WholeStageCodegen表示全阶段代码生成性能最优Exchange表示Shuffle是性能瓶颈。注意collect()会把所有数据拉到Driver10GB数据必OOM。必须用take(10)或show(10)预览生产代码禁用collect()。避坑案例某用户行为分析需关联用户表1亿行与事件表50亿行。最初user_df.join(event_df, user_id)Shuffle数据量达2TB耗时3小时。改用broadcastJoinuser_df.join(broadcast(event_df), user_id)将小表广播到各Executor耗时降至11分钟。前提是小表10GBSpark默认阈值。3.10 Docker环境一致性的终极保险Docker不是虚拟机而是Linux内核的命名空间namespace与控制组cgroup封装。它隔离进程、网络、文件系统但共享宿主机内核启动秒级。关键参数与选择逻辑Dockerfile中COPY requirements.txt .后RUN pip install -r requirements.txt而非COPY . .后RUN pip install -r requirements.txt。前者利用Docker层缓存依赖不变时跳过安装。docker run -v $(pwd)/data:/app/data:ro挂载数据目录为只读:ro防止容器内误删原始数据。--memory4g --cpus2限制容器资源避免单个容器吃光宿主机内存。实操心得提示用.dockerignore排除__pycache__/,.git/,data/减小镜像体积。我们镜像从1.2GB降至320MB。注意CMD [python, app.py]是容器主进程它退出容器即停止。用supervisord管理多进程是反模式应一个容器一个主进程。避坑案例某模型服务容器在K8s中频繁重启。kubectl logs显示Killed但无错误。kubectl describe pod发现OOMKilled。排查发现requirements.txt中tensorflow未指定版本Docker build时安装了2.12内存占用翻倍。解决方案固定tensorflow2.11.0并用docker system df -v监控镜像大小。3.11 MLflow机器学习生命周期的记录仪MLflow不是模型托管平台而是实验跟踪Tracking、项目打包Projects、模型注册Models的元数据记录仪。它不运行代码只记录谁、何时、用什么参数、得到什么指标。关键参数与选择逻辑mlflow.set_tracking_uri(http://mlflow-server:5000)指向中央服务器而非本地file:///mlruns。with mlflow.start_run(run_namerf_tuning_v2):run_name必须唯一便于搜索。我们用{model}_{params_hash}_{timestamp}格式。mlflow.log_artifact(model.pkl)记录模型文件但大文件100MB应存S3只记录URI。实操心得提示用mlflow.sklearn.autolog()自动记录Scikit-learn训练参数、指标、模型。但需mlflow.start_run()后调用否则无效。注意mlflow.search_runs()返回DataFrame可直接用Pandas分析。runs mlflow.search_runs(filter_stringmetrics.auc 0.85)筛选高分实验。避坑案例某次模型迭代5个工程师各自mlflow.start_run()产生200实验无法对比。我们推行“实验即分支”每个Feature分支对应一个run_nameCI流程中自动注入GIT_COMMIT_SHA作为run_tag用mlflow.search_runs(filter_stringtags.git_commit abc123)一键追踪。3.12 GitOpsArgo CD声明式部署的守门人GitOps不是CI/CD工具而是将Git仓库作为唯一事实源Source of Truth通过自动化同步确保集群状态与声明一致。Argo CD是其实现。关键参数与选择逻辑ApplicationCRD中spec.source.path: manifests/prod指定K8s YAML存放路径而非.避免同步无关文件。syncPolicy.automated.prunetrue自动删除YAML中已移除的资源但selfHealfalse禁用自动修复避免误删。health.lua自定义健康检查脚本判断Deployment是否Ready。实操心得提示用argocd app sync my-app --prune --force手动同步--prune删除多余资源--force跳过健康检查紧急修复用。注意argocd app wait my-app --health --sync等待应用健康且同步完成CI中必须用此命令确保部署成功。避坑案例某次发布Argo CD同步后Pod Running但服务不可用。argocd app health显示Progressing因health.lua未定义。我们为所有应用添加标准健康检查if obj.status ~ nil and obj.status.conditions ~ nil then for _, c in ipairs(obj.status.conditions) do if c.type Available and c.status True then return Healthy end end end return Progressing。4. 工具链协同实战一个完整的用户流失预测项目4.1 项目背景与数据流设计我们为一家SaaS公司构建用户流失预测模型目标是提前30天识别高风险用户准确率85%。数据源包括行为日志Kafka实时流JSON格式每秒10万事件存入S3s3://logs/raw/2022/12/01/用户档案PostgreSQL表users120万行含注册时间、套餐等级、付费状态产品使用Snowflake表feature_usage每日增量10GB数据流设计为三层接入层Flink消费Kafka清洗后写入S3 Parquet分区dt20221201处理层PySpark读取S3 Parquet PostgreSQL Snowflake特征工程生成宽表user_features_20221201.parquet建模层Dask加载宽表训练XGBoost模型MLflow记录实验模型存入S3s3://models/churn/v20221201/服务层Flask API加载模型Docker容器化Argo CD部署到K8s关键决策不用Airflow调度因Flink已处理实时流不用Kubeflow因团队熟悉DockerK8sMLflow不托管模型因S3权限体系更成熟。4.2 各环节工具选择与参数详解特征工程PySpark# 读取S3日志Parquet logs_df spark.read.parquet(s3a://logs/processed/dt20221201/) \ .filter(event_type IN (login, click, pay)) \ .withColumn(dt, to_date(event_time)) # 关联用户表PostgreSQL users_df spark.read \ .format(jdbc) \ .option(url, jdbc:postgresql://pg:5432/db) \ .option(dbtable, users) \ .option(user, user) \ .option(password, pass) \ .load() # 特征聚合过去7天登录次数、最近付费时间 features_df logs_df \ .filter(col(dt) date_sub(current_date(), 7)) \ .groupBy(user_id) \ .agg( count(when(col(event_type) login, 1)).alias(login_7d), max(when(col(event_type) pay, col(event_time))).alias(last_pay) ) # 写入S3宽表