Positron 现在可以运行在 Amazon SageMaker AI 环境中。对数据科学团队而言，真正有价值的并不是多了一个 IDE，而是可以在同一个受治理的 SageMaker Studio Space 里完成数据探索、特征验证、模型训练、实时部署和结果发布，同时继续使用 R、Python 与 Quarto 各自擅长的工具链。
下面用一条可改造的工作流说明如何组织这样的项目。示例假设 Athena 表已经存在，Studio Space 的执行角色拥有读取 Athena、访问指定 S3 路径以及创建 SageMaker Endpoint 的权限。具体镜像、Positron 配置和可用框架版本需要根据所在区域与组织环境调整。
一个项目，不必强迫团队只用一种语言
典型的数据科学项目往往跨越多个运行时：
- Athena 负责查询存放在数据湖中的明细数据；
- R 用于快速检查变量分布、缺失率和异常值；
- Python 与 XGBoost 承担模型训练和评估；
- SageMaker Endpoint 提供在线推理；
- Quarto 把指标、图表和结论编译成可审阅报告。
Positron 的作用是把这些工作放进同一个开发界面，而 SageMaker Studio Space 则提供持久化工作目录、身份权限和基础设施边界。项目可以按下面的方式组织：
customer-risk/
├── data/
├── models/
├── reports/
│ └── model-report.qmd
├── src/
│ ├── extract_athena.py
│ ├── validate_features.R
│ ├── train.py
│ ├── inference.py
│ └── deploy.py
└── requirements.txt
建议不要把访问密钥、数据库密码或账号标识写进项目文件。Studio 中的代码应通过执行角色取得临时凭证，并用 IAM 策略限制能够访问的 Athena Workgroup、Glue Data Catalog、S3 前缀和 SageMaker 资源。
从 Athena 拉取可复现的数据切片
下面的脚本使用 AWS SDK for pandas 查询 Athena，并把结果保存成 Parquet。运行前需要修改数据库、表名、Workgroup 和 Athena 查询结果所在的 S3 路径。
# requirements.txt
awswrangler>=3.9,<4
pandas>=2.1,<3
pyarrow>=15,<20
xgboost>=2.0,<3
scikit-learn>=1.4,<2
sagemaker>=2.220,<3
# src/extract_athena.py
from pathlib import Path
import awswrangler as wr
DATABASE = "analytics"
TABLE = "customer_features"
WORKGROUP = "data-science"
ATHENA_OUTPUT = "s3://replace-me-athena-results/queries/"
sql = f"""
SELECT
customer_id,
account_age_days,
orders_30d,
spend_30d,
support_tickets_30d,
churned
FROM {TABLE}
WHERE snapshot_date = current_date - interval '1' day
AND account_age_days IS NOT NULL
"""
frame = wr.athena.read_sql_query(
sql=sql,
database=DATABASE,
workgroup=WORKGROUP,
s3_output=ATHENA_OUTPUT,
ctas_approach=True,
)
output = Path("data/customer_features.parquet")
output.parent.mkdir(parents=True, exist_ok=True)
frame.to_parquet(output, index=False)
print(f"wrote {len(frame):,} rows to {output}")
在 Positron 的终端中执行：
python -m pip install -r requirements.txt
python src/extract_athena.py
生产环境还应把查询文本、数据快照日期和输出对象版本写入运行记录。否则，即使训练代码没有变化，也可能因为上游表发生更新而无法复现实验。
用 R 在训练前拦住坏特征
R 很适合把数据质量检查写成简短、可失败的脚本。下面的示例检查标签范围、重复主键、缺失率和明显不合理的负数。假设本项目中的计数及金额特征不应小于零。
# src/validate_features.R
library(arrow)
library(dplyr)
features <- read_parquet("data/customer_features.parquet")
required <- c(
"customer_id", "account_age_days", "orders_30d",
"spend_30d", "support_tickets_30d", "churned"
)
missing_columns <- setdiff(required, names(features))
stopifnot(length(missing_columns) == 0)
stopifnot(all(features$churned %in% c(0, 1)))
stopifnot(n_distinct(features$customer_id) == nrow(features))
numeric_features <- features |>
select(account_age_days, orders_30d, spend_30d, support_tickets_30d)
missing_rate <- sapply(numeric_features, function(x) mean(is.na(x)))
print(missing_rate)
stopifnot(all(missing_rate <= 0.05))
stopifnot(all(numeric_features$account_age_days >= 0, na.rm = TRUE))
stopifnot(all(numeric_features$orders_30d >= 0, na.rm = TRUE))
stopifnot(all(numeric_features$spend_30d >= 0, na.rm = TRUE))
stopifnot(all(numeric_features$support_tickets_30d >= 0, na.rm = TRUE))
cat("feature validation passed\n")
安装依赖并运行：
R -e 'install.packages(c("arrow", "dplyr"), repos="https://cloud.r-project.org")'
Rscript src/validate_features.R
这些阈值只是可改造的示例。真实项目应根据业务语义设置范围，并明确缺失值是数据错误、业务状态，还是需要由模型处理的有效信号。
在 Python 中训练 XGBoost，并部署实时端点
训练脚本只使用数值特征，把评估结果写入 JSON，并将模型保存为 XGBoost 原生格式。分层切分可以避免小样本场景下测试集标签比例严重偏移。
# src/train.py
import json
from pathlib import Path
import pandas as pd
import xgboost as xgb
from sklearn.metrics import accuracy_score, roc_auc_score
from sklearn.model_selection import train_test_split
FEATURES = [
"account_age_days",
"orders_30d",
"spend_30d",
"support_tickets_30d",
]
TARGET = "churned"
df = pd.read_parquet("data/customer_features.parquet").dropna(
subset=FEATURES + [TARGET]
)
X_train, X_test, y_train, y_test = train_test_split(
df[FEATURES],
df[TARGET],
test_size=0.2,
random_state=42,
stratify=df[TARGET],
)
model = xgb.XGBClassifier(
n_estimators=300,
max_depth=5,
learning_rate=0.05,
subsample=0.8,
colsample_bytree=0.8,
eval_metric="logloss",
random_state=42,
)
model.fit(X_train, y_train)
probability = model.predict_proba(X_test)[:, 1]
prediction = (probability >= 0.5).astype(int)
metrics = {
"rows": int(len(df)),
"test_rows": int(len(X_test)),
"roc_auc": float(roc_auc_score(y_test, probability)),
"accuracy": float(accuracy_score(y_test, prediction)),
}
Path("models").mkdir(exist_ok=True)
model.save_model("models/xgboost-model")
Path("models/metrics.json").write_text(
json.dumps(metrics, indent=2), encoding="utf-8"
)
print(json.dumps(metrics, indent=2))
python src/train.py
部署时需要一个推理入口。以下实现接收 CSV，每行按照 FEATURES 中的固定顺序提供四个数值，并返回流失概率：
# src/inference.py
import io
import json
import os
import numpy as np
import xgboost as xgb
def model_fn(model_dir):
model = xgb.XGBClassifier()
model.load_model(os.path.join(model_dir, "xgboost-model"))
return model
def input_fn(body, content_type):
if content_type != "text/csv":
raise ValueError(f"unsupported content type: {content_type}")
values = np.loadtxt(io.StringIO(body), delimiter=",")
return np.atleast_2d(values)
def predict_fn(data, model):
return model.predict_proba(data)[:, 1]
def output_fn(prediction, accept):
return json.dumps({"probabilities": prediction.tolist()}), "application/json"
接着将模型打包、上传到 S3，并创建实时端点。修改桶名、端点名、实例类型和 framework_version；后者必须与训练时的 XGBoost 版本及所在区域支持的 SageMaker 容器兼容。
tar -C models -czf model.tar.gz xgboost-model
export MODEL_BUCKET="replace-me-model-bucket"
aws s3 cp model.tar.gz "s3://${MODEL_BUCKET}/customer-risk/model.tar.gz"
# src/deploy.py
import os
import sagemaker
from sagemaker.serializers import CSVSerializer
from sagemaker.deserializers import JSONDeserializer
from sagemaker.xgboost.model import XGBoostModel
session = sagemaker.Session()
role = sagemaker.get_execution_role()
model_uri = f"s3://{os.environ['MODEL_BUCKET']}/customer-risk/model.tar.gz"
model = XGBoostModel(
model_data=model_uri,
role=role,
entry_point="inference.py",
source_dir="src",
framework_version="2.1-1",
py_version="py3",
sagemaker_session=session,
)
predictor = model.deploy(
initial_instance_count=1,
instance_type="ml.m5.large",
endpoint_name="customer-risk-xgb",
)
predictor.serializer = CSVSerializer()
predictor.deserializer = JSONDeserializer()
print(predictor.predict([[365, 3, 249.90, 1]]))
python src/deploy.py
实时端点会持续计费。完成测试后应显式删除：
aws sagemaker delete-endpoint --endpoint-name customer-risk-xgb
示例使用本地训练来突出 Positron 中的跨语言工作流。数据量增大后，可以保留相同项目结构，但把训练步骤迁移到 SageMaker Training Job，让计算资源、日志和模型产物与交互式开发环境解耦。
用 Quarto 把模型结果变成可审阅文档
Quarto 报告可以直接读取训练生成的 metrics.json。这样报告中的数字来自实际运行产物，而不是人工复制。
---
title: "Customer risk model report"
format: html
execute:
echo: true
---
## Evaluation
```{python}
import json
from pathlib import Path
metrics = json.loads(Path("../../models/metrics.json").read_text())
print(f"Rows: {metrics['rows']:,}")
print(f"ROC AUC: {metrics['roc_auc']:.3f}")
print(f"Accuracy: {metrics['accuracy']:.3f}")
Deployment decision
Deploy only after feature validation passes and the model meets the approved threshold.
在项目根目录执行：
```bash
quarto render reports/model-report.qmd
如果报告要作为正式审批材料，还应补充数据快照、代码提交号、训练参数、阈值选择依据、偏差检查和端点版本，而不只是展示一个 ROC AUC。
落地时重点检查什么
把 Positron 放入 SageMaker Studio 并不自动等于工作流已经受治理。上线前至少检查以下项目：
- 权限最小化：Space 执行角色只允许访问必要的数据库、S3 前缀和 SageMaker API。
- 依赖可复现：锁定 Python、R 和系统依赖；训练容器与推理容器中的 XGBoost 版本保持兼容。
- 数据可追溯：记录 Athena SQL、快照日期、Glue 表版本和输出对象位置。
- 开发与生产分离：探索可在 Space 中进行，正式训练和部署应由可审计的流水线触发。
- 成本有退出机制：为实时端点设置标签、告警和自动清理流程。
- 报告不代替验证：Quarto 适合传播结果，但模型审批仍需覆盖偏差、漂移、安全与业务风险。
这套组合的优势不是让 R 和 Python 互相替代，而是让它们在同一个受控项目中各司其职。先用一个低风险数据集验证权限、依赖和端点清理流程，再逐步接入正式数据与自动化流水线，会比一次性迁移整个数据科学平台稳妥得多。