Spork: Pig on Spark 實現深度分析：物理算子到 RDD 轉換、Launcher 架構、核心算子映射表

項目背景

Spork 是 Apache Pig 在 Apache Spark 上的實驗性運行時（Pig on Spark）。將 Pig 的物理執行計劃（Physical Plan）翻譯為 Spark RDD 操作，繞過 MapReduce 階段。本文基於 flare-spork 分支分析。

架構概覽：SparkLauncher

對應 MR 模式的 MapReduceLauncher，核心職責：

• 接收物理運算符鏈（PhysicalOperator 子類實例）
• 按拓撲順序遍歷，調用對應 POConverter 轉換為 RDD 運算
• 最終觸發 Action（Store/Dump）提交 Spark Job

Converter 接口定義

public interface POConverter {
    RDD convert(List> predecessorRdds, T physicalOperator) throws IOException;
}

• predecessorRdds：上游算子產生的 RDD 列表（對應物理計劃的邊）
• 返回 nextRDD：下游算子的輸入
• 缺點：直接映射無優化，整個物理計劃可能對應多個 Spark Job（每個 Action 觸發一次）

核心算子轉換映射表

關鍵算子轉換細節

POForEach：Iterator 模式復用 Pig 邏輯

// 核心：實現 Iterator => Iterator
rdd.mapPartitions(iter -> {
    return new Iterator() {
        POForEach po = (POForEach) physicalOperator;
        Iterator inputIter = iter;
        
        public boolean hasNext() {
            if (!inputIter.hasNext()) return false;
            Tuple t = inputIter.next();
            po.setInput(t);        // 設置輸入
            po.attachInput(t);     // 附加輸入
            return po.getNextTuple() != null;
        }
        public Tuple next() {
            return po.getNextTuple(); // 觸發 processTuple()
        }
    };
});

關鍵：Pig 物理算子通過 setInput/attachInput 接收數據，getNextTuple 觸發處理邏輯。Converter 在 Iterator 中模擬此協議。

POGlobalRearrange：分區策略

• 單父 RDD：rdd.groupByKey() → map 取 (key, Iterator)
• 多父 RDD：各父 RDD 先 map 為 (key, value) → new CoGroupRDD → map 合併 Iterator
• CoGroupRDD 內部對同 Key 的多個 Iterator 做 merge（類似歸並）

POSort：有序分區

// 1. Tuple -> Tuple2
rdd.map(t -> new Tuple2<>(t, null))
// 2. OrderedRDDFunctions.sortByKey(ascending, numPartitions)
//    內部使用 RangePartitioner，各分區內有序，分區間按 Key 範圍分布
// 3. mapPartitions 取回 Key
sorted.mapPartitions(iter -> iter.map(t2 -> t2._1))

比較器復用 POSort.mComparator，保證排序語義一致。

啟動方式與局限性

啟動命令

pig -x spark script.pig
# 或
java -cp ... org.apache.pig.Main -x spark script.pig

已知局限

1. 無 Catalyst 優化：直接物理算子→RDD 映射，缺少邏輯計劃優化（谓詞下推、列剪枝、Join Reorder）
2. Job 邊界由 Action 決定：每個 Store/Dump 觸發一次 Spark Job，可能產生過多 Stage
3. Shuffle 依賴 Spark 默認：Join、GlobalRearrange 走 HashPartitioner/RangePartitioner，無自定義分區優化
4. 實驗性質：生產環境建議用 Pig on Tez 或 Spark SQL

總結

Spork 展示了「傳統 DAG 引擎（Pig）向新一代 DAG 引擎（Spark）遷移」的典型路徑：

1. Launcher 替換：保留前端解析、優化器，替換後端執行
2. 算子一一映射：定義 Converter 接口，逐個實現 PhysicalOperator → RDD Op
3. 協議適配：在 Iterator/Function 中模擬 Pig 算子的 setInput/getNextTuple 生命週期
4. 分區語義保持：Local/Global Rearrange、Sort、Package 對應 Spark 的 mapPartitions/groupByKey/sortByKey

雖 Spork 未成生產級，但其架構設計為理解 Pig、Spark 內核、跨引擎遷移提供了極佳案例。