第5–7次实验：集群、Cholesky分解与Spark机器学习

1.镜像、容器与分布式查询

课程原文：Docker使用操作系统级虚拟化，把应用及其依赖放入容器。镜像包含代码、库、工具和依赖。容器是镜像的运行实例。第5次实验使用Docker部署Hadoop、Spark和Drill。实验使用Docker Swarm的覆盖网络连接运行Docker守护进程的多台主机。

• 镜像（image）：只读的分层模板。一个镜像可以创建多个容器。
• 容器（container）：镜像的运行实例。容器具有独立的进程视图和可写层。
• Swarm管理节点（manager）：初始化Swarm。管理节点签发加入令牌。管理节点维护节点成员关系。
• 工作节点（worker）：使用管理节点给出的令牌和地址加入Swarm。
• 覆盖网络：使不同Docker主机上的容器进入同一个逻辑二层网络。

可验证推导：容器能够启动，只能说明本机容器运行时可用。容器通过名称访问另一台主机上的容器，还依赖Swarm成员关系、覆盖网络和网络端口。必须按层检查故障。

1. 运行 docker ps -a。检查容器状态。
2. 在管理节点运行 docker node ls。检查Swarm节点。
3. 运行 docker network ls。检查 hadoop-net。
4. 进入容器。测试名称解析和 ping。
5. 检查应用进程、端口和日志。

2.第5次实验的启动顺序和检查项

课程原文：管理节点和工作节点必须先能互相 ping。管理节点初始化Swarm。管理节点创建可附加的 hadoop-net。工作节点使用生成的令牌加入Swarm。然后启动Hadoop容器、Hadoop守护进程、ZooKeeper和Drillbit。

管理节点

docker system prune
master/init_swarm.sh
master/init_network.sh
docker node ls
docker network ls

export WORKER_NUMBER=N
master/start.sh
master/start_hadoop.sh
jps
./run-wordcount.sh

每个工作节点

docker system prune
worker/join_swarm.sh <JOIN_TOKEN> <MANAGER_IP>

export WORKER_NUMBER=N
export WORKER_ID=X
worker/start.sh
docker ps -a

需要区分：N 表示工作节点总数。X 表示当前工作节点的编号。源材料没有规定所有集群都使用同一个具体数值。

风险提示：docker system prune 会删除未使用的Docker对象。课程步骤包含该命令，但真实环境中应先检查待删除对象；不要把清理命令当作无副作用的“重启”。

3. Hadoop四层角色

课程原文：第5次实验在主容器中启动 NameNode、DataNode、ResourceManager 和 NodeManager。课程的Hadoop材料进一步区分HDFS、YARN和MapReduce。

hdfs dfs -mkdir /input
hdfs dfs -put local.csv /input
hdfs dfs -ls /input
hdfs dfs -cat /input/local.csv
hdfs dfs -get /output ./local_output

需要区分：HDFS路径不是本机路径。hdfs dfs -put 从本机复制文件到HDFS。-get 从HDFS复制文件到本机。Hadoop Streaming的输出目录必须事先不存在。

hadoop jar "$HADOOP_HOME/share/hadoop/tools/lib/hadoop-streaming-3.5.0.jar" \
  -files mapper.py,reducer.py \
  -input /user/hadoop/input \
  -output /user/hadoop/output \
  -mapper mapper.py \
  -reducer reducer.py

4. Drill：读取数据时确定模式的分布式SQL

课程原文：Apache Drill是面向大数据探索的低延迟分布式SQL查询引擎，可查询JSON、Parquet、CSV与Avro等数据。分布式模式中，Drillbit是查询执行服务，ZooKeeper用于协调与服务发现。

./build/docker-build-drill-image.sh
drill/start_zookeeper.sh
drill/start_drillbit.sh 1

# 同一主机再启动一个 Drillbit 时，宿主机端口必须不同
DRILL_HTTP_PORT=8048 DRILL_USER_PORT=31020 drill/start_drillbit.sh 2

docker exec hadoop-drillbit-1 /usr/local/drill/bin/sqlline \
  -u jdbc:drill:drillbit=localhost \
  --outputformat=tsv --silent=true \
  -e 'SELECT * FROM sys.drillbits;'

课程原文：在 sys.drillbits 的正常示例中，各Drillbit的状态为 ONLINE。恰有一个节点的 current 为 true，表示当前连接的Drillbit。该值不表示集群领导节点。上面的命令在容器 hadoop-drillbit-1 内通过 jdbc:drill:drillbit=localhost 建立直连，所以运行实例是 hadoop-drillbit-1。如果使用ZooKeeper连接字符串，则通过查询结果中 current=true 的行识别实际实例。

CSV的显式字段解释

SELECT *
FROM (
  SELECT
    columns[0] AS name,
    columns[1] AS id,
    CAST(columns[2] AS INT) AS grade
  FROM dfs.`/path/grades.csv`
)
WHERE name = 'Tad Wyze'
ORDER BY grade DESC;

需要区分：读取时确定模式（schema-on-read）不表示所有字段会自动获得正确类型。文本CSV的 columns[n] 初始为字符串；执行数值排序或聚合前，应显式使用 CAST 转换为数值类型。

5.三道航班Drill题：仓库中的实际SQL

仓库事实：labs/l5/src/lab5_drill.sql 查询HDFS上的 dfs.root.`/user/hadoopuser/lab5/flights_parquet`。实际列名是 UniqueCarrier、DepDelay、Origin 与 WeatherDelay；第一题按正的出发延误次数排名，第二题按天气延误分钟总和排名，第三题求各承运人的最大出发延误。

-- 1：正的出发延误次数最多的承运人（仓库保留前 10）
SELECT UniqueCarrier, COUNT(*) AS late_count
FROM dfs.root.`/user/hadoopuser/lab5/flights_parquet`
WHERE DepDelay > 0
GROUP BY UniqueCarrier
ORDER BY late_count DESC
LIMIT 10;

-- 2：天气延误分钟总和最大的三个出发地
SELECT Origin, SUM(WeatherDelay) AS weather_delay
FROM dfs.root.`/user/hadoopuser/lab5/flights_parquet`
WHERE WeatherDelay > 0
GROUP BY Origin
ORDER BY weather_delay DESC
LIMIT 3;

-- 3：每个承运人的最大出发延误
SELECT UniqueCarrier, MAX(DepDelay) AS max_dep_delay
FROM dfs.root.`/user/hadoopuser/lab5/flights_parquet`
GROUP BY UniqueCarrier
ORDER BY UniqueCarrier;

仓库结果：第一题首位是 DL（8,064,705次）；第二题前三名为 ATL、DFW、ORD，对应天气延误总和2,875,797、2,420,428、1,981,125。注意“最常晚点”也可被解释为晚点比例，但本仓库明确实现的是 DepDelay > 0 的次数。

6.故障定位清单

• 工作节点无法加入：先测试管理节点的传输控制协议（TCP）端口 2377。课程材料建议允许 2377。覆盖网络还需要 7946 和 4789。
• Drillbit退出：运行 docker ps --filter name=hadoop-drillbit 和 docker logs hadoop-drillbit-1。
• Drill内存不足：课程示例用 DRILL_HEAP=512M DRILL_MAX_DIRECT_MEMORY=1G 降低限制后启动。
• ZooKeeper异常：运行 docker ps --filter name=hadoop-zookeeper 与相应 docker logs。

可验证推导：宿主机端口映射冲突和容器内部服务端口属于不同问题。在同一台主机运行多个Drillbit时，必须使用不同的宿主机端口。不同主机可以使用相同的宿主机端口号。

1.为什么分解，不直接求逆

课程原文：数值线性代数实现通过矩阵分解求解线性系统 Ax = b，不显式计算 A⁻¹。先做分解的代价是 O(n³)；同一 A 对多个右端项求解时，每次三角求解只需 O(n²)。

若 A = LU，则先解 Ly = b（前代），再解 Ux = y（回代）。Doolittle约定 L 的对角线全为1，从而固定LU缩放自由度。

2. SPD的判定与反例

定义：实矩阵 A 是对称正定矩阵（symmetric positive definite，SPD），当且仅当：

课程原文：对实对称矩阵，以下两种方法可检查正定性：

1. 全部特征值严格大于0；
2. Sylvester判据：从左上角开始的所有顺序主子式严格大于0。

可验证推导：若 A = LL^T 且 L 可逆，则对 x ≠ 0，

因此正对角的非奇异Cholesky因子直接证明 A 正定。

需要区分：

• A = A^T 只证明对称，不能证明正定。例如 diag(1, -1) 对称但不定。
• 所有对角元素为正也不够。例如 [[1,2],[2,1]] 的行列式为 −3，不正定。
• A^TA 总是半正定；只有当 A 满列秩时才正定。

3.从Doolittle到LDLᵀ，再到Cholesky

课程原文与可验证推导：以下等式逐步给出Cholesky的来源。

1. 从Doolittle分解开始：A = LU，其中 L 为单位下三角，U 为上三角。
2. 因为 A 对称，LU = A = A^T = U^TL^T。
3. 把 U 的对角抽出：U = DÛ，其中 D = diag(U₁₁,…,U_nn)，Û 为单位上三角。
4. 于是 LDÛ = Û^TDL^T。Doolittle分解的唯一性给出 L = Û^T。
5. 因此 A = LDL^T。
6. 若 A 为正定，则每个枢轴 D_ii > 0，实数平方根 D^1/2 存在。
7. 令 C = LD^1/2，得到 A = CC^T。后续公式把 C 重新记作 L。

需要区分：某些非正定对称矩阵仍可能存在 LDL^T 分解。只有从 D 取实平方根并得到标准Cholesky分解时，才需要正定条件。

4.从矩阵乘法推导元素公式

由 A = LL^T，对任意 i,j：

先计算第 j 列的对角元。再计算该对角元下方的元素：

def cholesky(A):
    n = len(A)
    L = [[0.0] * n for _ in range(n)]
    for j in range(n):
        diagonal_residual = A[j][j] - sum(L[j][k] ** 2 for k in range(j))
        if diagonal_residual <= 0:
            raise ValueError("matrix is not SPD")
        L[j][j] = diagonal_residual ** 0.5
        for i in range(j + 1, n):
            cross = sum(L[i][k] * L[j][k] for k in range(j))
            L[i][j] = (A[i][j] - cross) / L[j][j]
    return L

可验证推导：循环不变量如下：开始计算第 j 列之前，L 的前 j−1 列已经满足对应的 LL^T 元素等式。对角残差必须严格大于0。

5. 3×3完整手算

课程原文：

1. l₁₁ = √25 = 5
2. l₂₁ = 15/5 = 3，l₃₁ = −5/5 = −1
3. l₂₂ = √(18−3²) = 3
4. l₃₂ = (0−(−1)(3))/3 = 1
5. l₃₃ = √(11−(−1)²−1²) = 3

验算：第 (2,3) 项为 3(−1)+3(1)+0(3)=0；第 (3,3) 项为 (−1)²+1²+3²=11。故 LL^T=A。

同一分解如何解方程

可验证推导：若取 b=[35,33,6]^T，则先解 Ly=b：

1. 5y₁=35 ⇒ y₁=7
2. 3y₁+3y₂=33 ⇒ y₂=4
3. −y₁+y₂+3y₃=6 ⇒ y₃=3

再解 L^Tx=y，得到 x=[1,1,1]^T。直接乘 Ax 可验证结果。

6.复杂度与适用范围

课程原文：Cholesky的主导运算量约为稠密LU的一半。它只计算一个三角因子，并利用对称性省去另一半。

1. Scala基本语法

课程原文：Scala是静态类型、多范式语言，可与Java互操作，同时支持面向对象和函数式编程。

val fixed: Int = 10                 // 不可重新赋值
var mutable: String = "Hello"       // 可以重新赋值
mutable = "World"

def add(a: Int, b: Int): Int = a + b
val doubled = List(1, 2, 3).map(_ * 2)
val total = List(1, 2, 3).reduce(_ + _)

case class Point(x: Int, y: Int)
val sameValue = Point(1, 2) == Point(1, 2)  // true

val description = fixed match {
  case 1 => "one"
  case _ => "other"
}

object HelloWorld {
  def main(args: Array[String]): Unit = {
    println("Hello, world!")
  }
}

scalac HelloWorld.scala
scala HelloWorld

2. Spark：驱动程序、RDD与DataFrame

课程原文：每个Spark应用都有驱动程序（driver program）。驱动程序运行用户的 main 方法，并组织集群上的并行操作。弹性分布式数据集（RDD）是可容错、分区的分布式元素集合。DataFrame是带模式的分布式表。Spark SQL使用Catalyst优化器处理DataFrame查询。

val textFile = spark.sparkContext.textFile("path/to/text.txt")
val wordCounts = textFile
  .flatMap(line => line.split(" "))
  .map(word => (word, 1))
  .reduceByKey(_ + _)

wordCounts.collect().foreach(println)

可验证推导：map 和 flatMap 等转换操作建立计算血缘。collect 等动作触发执行。collect 把全部结果传回驱动程序，所以只适用于能够装入驱动程序内存的结果。

需要区分：当聚合函数满足结合律并可在映射端合并时，reduceByKey 会在混洗前执行本地合并，因此比先执行 groupByKey 再求和传输更少的数据。

3. PBMC数据预处理

课程原文：Lab 7使用预处理后的PBMC mRNA表。数据有1882个特征；目标是用PCA降维，并用细胞类型标签做分类实验。

spark_df = (
    spark.read
    .options(header="true", inferSchema="true")
    .csv("datasets/PBMC_16k_RNA.csv")
)

spark_df.agg({"KLHL17": "max"}).show()
spark_df.createOrReplaceTempView("data")
spark.sql("SELECT max(KLHL17) FROM data").show()

课程原文：KLHL17 的范围约为 −0.11到10，HES4 的范围约为 −1.33到9.47。课程答案认为两者范围相近，可以跳过标准化，也可以为了稳妥执行标准化。

可验证推导：仅检查两个特征不足以证明全部1882个特征尺度相近。PCA对尺度敏感；严谨流程应统计所有输入列，或在含义允许时使用 StandardScaler。

from pyspark.ml.feature import VectorAssembler, StandardScaler

feature_cols = [c for c in spark_df.columns if c != "index"]
assembled = VectorAssembler(
    inputCols=feature_cols,
    outputCol="features"
).transform(spark_df)

scaler = StandardScaler(
    inputCol="features",
    outputCol="standardized_features",
    withMean=True,
    withStd=True
)
standardized_df = scaler.fit(assembled).transform(assembled)

未给定：源材料没有提供完整列清单，也没有确认索引列的精确名称。上例的 "index" 是占位名称。实际执行前必须检查数据模式，再确定需要排除的列。

4. PCA：目标、公式与课程结果

可验证推导：对中心化数据矩阵 X，协方差矩阵可写为 S = X^TX/(n−1)。第一主成分方向 w₁ 解约束优化：

拉格朗日条件给出 Sw = λw，因此主成分是协方差矩阵按特征值从大到小排列的特征向量。第 j 个解释方差比为 λ_j/Σλ_i。

from pyspark.ml.feature import PCA

pca = PCA(
    k=2,
    inputCol="standardized_features",
    outputCol="pca_features"
)
pca_model = pca.fit(standardized_df)
pca_df = pca_model.transform(standardized_df)
print(pca_model.explainedVariance.toArray())

课程原文：前两个解释方差比分别约为 0.02942326 与 0.01307317，合计约 0.04249643，即4.25%。课程答案随后写“2% 的方差”；这与前述两个数值之和不一致。复习时应优先按给定数值求和，并指出原文表述冲突。

课程原文：在本实验按标签着色的前两个主成分散点图中，同类细胞形成聚集区域；随机选择两个原始列时，同类细胞更分散。该对比说明PCA可用于本数据集的低维可视化。

需要区分：较低的解释方差不表示PCA计算错误。它只说明二维投影丢失了大部分总方差。分类效果还取决于被丢弃的方向是否包含分类信息。

5.逻辑回归、BFGS与L-BFGS

课程原文：第7次实验把样本分为A类和非A类，并要求使用两个主成分训练逻辑回归。课程计数结果显示 CD4+ T 数量最多，共5262个样本，因此把它定义为A类。训练集和测试集按70% 和30% 划分。

可验证推导：二分类逻辑回归令

• 梯度下降：只用一阶梯度，沿负梯度方向更新。
• Newton方法：使用Hessian矩阵求解步长。高维问题的存储和求解成本较高。
• BFGS方法：使用梯度差和步长更新逆Hessian近似，不显式计算真实Hessian矩阵。
• L-BFGS方法：只保留最近有限次 (s_k,y_k) 历史，以降低高维问题的内存成本。

from pyspark.mllib.classification import LogisticRegressionWithLBFGS
from pyspark.mllib.linalg import Vectors as OldVectors
from pyspark.mllib.regression import LabeledPoint

# 前置条件：classified_df 已含数值 label 与 ML Vector 类型的 pca_features
labeled_points = classified_df.select("label", "pca_features").rdd.map(
    lambda row: LabeledPoint(
        float(row["label"]),
        OldVectors.dense(row["pca_features"].toArray())
    )
)

training, test = labeled_points.randomSplit([0.7, 0.3], seed=42)

model = LogisticRegressionWithLBFGS.train(
    training,
    iterations=100,
    numClasses=2
)

prediction_and_label = test.map(
    lambda point: (model.predict(point.features), point.label)
)
test_error = prediction_and_label.filter(
    lambda pair: pair[0] != pair[1]
).count() / float(test.count())

课程原文：一次实验在100次迭代后得到25.59% 测试错误率。课程答案认为二维图上 CD4+ T 与其他类型很接近，并建议加入更多主成分。

未给定：25.59% 不是可复现的固定标准答案。源材料没有给出随机种子、精确预处理、特征组装、正则化参数和数据版本。结果应作为一次实验观测，而不是算法常数。

改进和验证步骤

1. 在验证集上选择主成分数 k。不要只比较训练误差。
2. 固定随机种子。报告训练样本量和测试样本量。
3. 标准化特征。只在训练集上拟合缩放器和PCA，以防止测试信息泄漏。
4. 调节正则化强度和迭代次数。检查目标函数是否收敛。
5. 类别不平衡时，同时报告精确率、召回率、F1值、受试者工作特征曲线下面积（ROC-AUC）和混淆矩阵。不要只报告准确率。
6. 如果二维线性边界不足，则增加主成分数，或比较非线性模型。

需要区分：先使用全部数据拟合PCA，再切分训练集和测试集，会使测试集参与特征空间学习。PCA不使用标签，但该做法仍然造成数据泄漏。