From SQL jams to pipeline style php

從 SQL 醬缸轉換為 pipeline 式的 PHP

在處理大量複雜訂單的操作時，常常混雜各式各樣的條件跟運作，例如:

A. 找出金額大於某個數字，並且下單時間在某個優惠時段的訂單，予以 N % 的紅利點數
B. 找出下單時間在某個優惠時段的訂單，予以 N % 的折扣
C. 找出金額大於某個數字，並且排除某個優惠時段的訂單，將訂單總金額某上某個比率，計入該消費者的總積分
…

依一般最常見的 PHP + MySQL 的作法，若有 100 條抓資料的操作，大概就會寫成 100 份(或更多) SQL 語法指令，並依對應的 MySQL 條件式組合。於是便會有一堆 WHERE / AND / OR / GROUP BY / HAVING / subquery / … 等交錯的組合結構。

在此，我先將這稱之為 SQL-driven 的開發方式。

因為 SQL 的語法指令本身不易模組化。所以兩個類似的操作，常常就是從某一個 SQL 指令複製貼上到另一個，再小改一部份。這種開發程式的方式，在初期很快很方便，能迅速增加功能跟解決眼前的小任務。但隨操作的種類增加、需求多元變化，複雜度不斷往上疊加後，漸漸會遇到下述困境:

其中一處抓資料的 SQL + PHP 指令組合有修改。從該處複製衍生出去的指令程式，也都要跟著修改。
但修改前後可能是不同工程師，後手不見得知道有那些地方要跟著更新，也不知那些地方不能改。
有改動到的地方，該功能很可能需要重新測試跟驗證。改越多地方，潛在的後座力越大。因此，多數人傾向只完成眼前任務所需的最少更改，而不會再進一步碰觸其他需要一併更新的 SQL + PHP 指令。
幾番迭代後，這堆 SQL + PHP 程式，很快就落入語法前後不一致，彼此相關，但又邏輯無法貫通的窘境。
中後期，程式碼維護越來越痛苦，開發過程中處處有地雷。
最終，在這程式碼中工作，就好像在玩疊疊樂(Jenga) ，每個人都在往上疊加複雜度跟風險。不時擔心會不會改了某段程式，就不小心踩到地雷而程式大爆炸。

會走入這種困境，其根本的原因之一在於，這種 SQL-driven 的開發方式，每個算法跟函式功能之間，不易重組利用。並非開發人員對 PHP 、 MYSQL 程式語言能力不夠，而是這種程式架構的發展方式本身有其侷限。加上很多專案開發只著重在不斷增加新功能，而幾乎沒有事後 code review 跟 refactor 的過程，很常會遇到這樣的困境。

那麼，除了上述 SQL-driven 外，是否還有其他的開發方式呢?

Pipeline style

在此，想介紹的另一種風格思維: Pipeline style。在過往的經歷中，曾受過啟發的地方如下:

Unix/Linux 上 Shell Script 普遍用到的 pipe ，例如:
- cat /proc/cpuinfo | grep vendor_id | cut -d: -f2 | uniq
  - 由 4 個獨立指令組成，中間透過 | 運算元的串接，前一指令的輸出會變成到下一指令的輸入
- 一個複雜的大任務，可分拆成小任務的組合，最終能用化約成個別獨立的指令的組合。
- 指令能組合並產生各式各樣的任務。
Hadoop 所提出的 MapReduce 的新作法。
MongoDB 的 Map-Reduce 方法
OpenVanilla 的 Method chaining:
- C++ 語法: buf->clear()->append((char*)"\xe7\x9a\x84")->update(); 來源
- Method chaining

在上述的啟發下，也跟著尋思在 PHP 上是否也能實踐這樣的架構。之後研究了一下，發現目前 PHP 裡有下述機制可以使用:

MapReduce 相關的 functions
- array_map()
- array_reduce()
- array_filter()
Anonymous functions ( v5.4 之後支援 )
- Anonymous functions 其實在 v5.3 時就已出現，但不完整
- v5.4 之後，始支援 object and class scope binding 。(註: Closure::bind ) 這之後才有辦法作到比較好的語法包裝。

於是便動手嘗試實作如下:

初步實作

假設有訂單的數據資料如下:

/*
 * Order states
 */
define("_NEW", 1);
define("_PENDING", 2);
define("_PROCESSING", 3);
define("_DONE", 4);

/*
 * Sample data
 */
$orders = array(
    ["order_id" => 1, "amount" => 1000, "bonus" => null, "date" => "2017-07-20T18:17:03+08:00", "state" => _PENDING],
    ["order_id" => 2, "amount" => 3000, "bonus" => null, "date" => "2017-07-21T21:38:24+08:00", "state" => _NEW],
    ["order_id" => 3, "amount" => 2500, "bonus" => null, "date" => "2017-07-22T09:41:51+08:00", "state" => _DONE],
    ["order_id" => 4, "amount" => 1800, "bonus" => null, "date" => "2017-07-23T13:24:31+08:00", "state" => _PENDING],
    ["order_id" => 5, "amount" => 1200, "bonus" => null, "date" => "2017-07-24T07:55:47+08:00", "state" => _DONE],
);

// Debug output
print_r($orders);

依 array_filter() 的特性，我們可以用來實作過濾函式如下:

依狀態(state)來過濾選取訂單

$filter_state = function ($state) {
    $res_func = function ($item) use ($state) {
        return $item["state"] == $state;
    };

    return $res_func;
};

// Apply the filter funtion on the order data
$result = array_filter($orders, $filter_state(_PENDING));

// Debug output
print_r($result);

依日期(date)來過濾選取訂單

/*
 * A filter function to filter by date
 */
$filter_date = function ($from, $to) {
    $res_func = function ($item) use ($from, $to) {
        return (strtotime($from) <= strtotime($item["date"]) && strtotime($item["date"]) <= strtotime($to));
    };

    return $res_func;
};

// Apply the filter funtion on the order data
$result = array_filter($orders, $filter_date("2017-07-21", "2017-07-23"));

// Debug output
print_r($result);

依 array_reduce() 的特性，我們可以用來實作計算函式如下:

計算資料集內某欄位的加總

/*
 * A function to calculate the sum of specified column
 */
$calculate_sum = function ($col) {
    $res_func = function ($carry, $item) use ($col) {
        $carry += $item[$col];
        return $carry;
    };
    return $res_func;
};

// Apply the calculate funtion on the order data
$result = array_reduce($orders, $calculate_sum("amount"));

// Debug output
print_r($result);

依 array_map() 的特性，我們可以用來實作附加函式如下:

計算每一條的 bonus 的點數

/*
 * A function to calculate the bonus point of the order
 */
$map_bonus =  function ($percent) {
    $res_func = function ($item) use ($percent) {
        $item["bonus"] += $item["amount"] * $percent / 100.0;
        return $item;
    };

    return $res_func;
};

// Apply the calculate funtion on the order data
$result = array_map($map_bonus(0.18), $orders);

// Debug output
print_r($result);

有了基本的操作組合之後，可依上述擬定的函式，可以將之組合成一連串的操作如下:

// filter by the date
$result1 = array_filter($orders, $filter_date("2017-07-21", "2017-07-23"));
// filter by the state
$result2 = array_filter($result1, $filter_state(_DONE));
// calculate the bonus points
$result3 = array_map($map_bonus(0.5), $result2);
// calculate the sum of bonus points
$result4 = array_reduce($result3, $calculate_sum("bonus"));

// Debug output
print_r(
    [
        "result1" => $result1,
        "result2" => $result2,
        "result3" => $result3,
        "result4" => $result4
    ]
);

略見雛型，但看起來沒什麼吸引力。

進一步改善: PipelineArray

首先，這樣的寫法少了 Method chaining 的特性，還不夠好用。於是，我進一步設計了一個 PipelineArray 的 Class 如下:

class PipelineArray
{
    public function __construct($rows)
    {
        $this->rows = $rows;
    }

    public function from_rows($rows)
    {
        $this->rows = $rows;
    }

    public function map($func)
    {
        $this->rows = array_map($func, $this->rows);
        return $this;
    }

    public function filter($func)
    {
        $this->rows = array_filter($this->rows, $func);
        return $this;
    }

    public function reduce($func)
    {
        $this->rows = array_reduce($this->rows, $func);
        return $this;
    }

    public function to_rows()
    {
        return $this->rows;
    }

    public function debug($msg)
    {
        print_r(
            [
                "msg" => "[DEBUG]: $msg",
                "rows" => $this->rows
            ]
        );
        return $this;
    }
}

以上述的 PipelineArray 包裝之後，程式的寫法可改善如下:

$result = (new PipelineArray($orders))
    ->filter($filter_date("2017-07-21", "2017-07-23"))
    ->filter($filter_state(_DONE))
    ->debug("step 1")
    ->map($map_bonus(0.5))
    ->debug("step 2")
    ->reduce($calculate_sum("bonus"))
    ->to_rows(); // Output in original row format

// Debug output
print_r(
    [
        "result" => $result,
    ]
);

改寫後，寫法看起來比較簡單明白一些。而有了 Method chaining 的元素之後，語法上也較容易重新組合功能操作函式。

進一步封裝: Order class

有了上述的基礎，我們可能再進一步封裝成 Order 的 class ，讓整體的操作更簡潔一點。像是:

繼承 PipelineArray class, 如此可重複使用先前的成果
將 NEW, PENDING, … 等狀態常數封裝進 Order
將語法 ->filter($filter_state(_DONE)) 精簡為 ->filter_state(_DONE)

回到訂單的部份，新架構的程式可改寫如下:

class Order extends PipelineArray
{

    const NEW = 1;
    const PENDING = 2;
    const PROCESSING = 3;
    const DONE = 4;

    public function filter_state($state)
    {
        return $this->filter(function ($item) use ($state) {
            return $item["state"] == $state;
        });
    }

    public function filter_date($from, $to)
    {
        return $this->filter(function ($item) use ($from, $to) {
            return (strtotime($from) <= strtotime($item["date"]) && strtotime($item["date"]) <= strtotime($to));
        });
    }

    public function calculate_sum($col)
    {
        return $this->reduce(function ($carry, $item) use ($col) {
            $carry += $item[$col];
            return $carry;
        });
    }

    public function map_bonus($percent)
    {
        return $this->map(function ($item) use ($percent) {
            $item["bonus"] += $item["amount"] * $percent / 100.0;
            return $item;
        });
    }

    public function getDoneBonusSum()
    {
        return $this
            ->filter_state(self::DONE)
            ->calculate_sum("bonus");
    }

    public function sample_run()
    {
        return $this
            ->filter_date("2017-07-21", "2017-07-23")
            ->map_bonus(0.5)
            ->getDoneBonusSum();
    }
}

使用時語法如下:

print_r(
    [
        "result" => (new Order($orders))->sample_run()->to_rows()
    ]
);

注意到，上述中的 sample_run() 的函式中，重複使用了 getDoneBonusSum() 的功能。而 getDoneBonusSum() 跟 sample_run() 也都是由先前的 filter_state(), filter_date(), calculate_sum(), map_bonus(), … 所疊加組合而成，都是未來可持續重複使用跟重組的函式。依照這個方向，可再持續實作更多的基礎操作函式，並產生更多相互疊加組合的功能。逐一涵蓋所有訂單相關的資料操作。

至此，大致利用了 PHP 現有的機制，完成了一個 Pipeline style 的初步架構設計。

若能掌握了這個方向，在遇到不斷發膨脹成長的 SQL + PHP 的程式困境時，可嘗試將底層的 SQL-driven 程式碼，逐步重構改為 Pipeline style 的實作，慢慢往可重複使用的架構收斂。

附註

最終 sample 程式碼如下:

<?php


class PipelineArray
{
    public function __construct($rows)
    {
        $this->rows = $rows;
    }

    public function from_rows($rows)
    {
        $this->rows = $rows;
    }

    public function map($func)
    {
        $this->rows = array_map($func, $this->rows);
        return $this;
    }

    public function filter($func)
    {
        $this->rows = array_filter($this->rows, $func);
        return $this;
    }

    public function reduce($func)
    {
        $this->rows = array_reduce($this->rows, $func);
        return $this;
    }

    public function to_rows()
    {
        return $this->rows;
    }

    public function debug($msg)
    {
        print_r(
            [
                "msg" => "[DEBUG]: $msg",
                "rows" => $this->rows
            ]
        );
        return $this;
    }
}

class Order extends PipelineArray
{

    const NEW = 1;
    const PENDING = 2;
    const PROCESSING = 3;
    const DONE = 4;

    public function filter_state($state)
    {
        return $this->filter(function ($item) use ($state) {
            return $item["state"] == $state;
        });
    }

    public function filter_date($from, $to)
    {
        return $this->filter(function ($item) use ($from, $to) {
            return (strtotime($from) <= strtotime($item["date"]) && strtotime($item["date"]) <= strtotime($to));
        });
    }

    public function calculate_sum($col)
    {
        return $this->reduce(function ($carry, $item) use ($col) {
            $carry += $item[$col];
            return $carry;
        });
    }

    public function map_bonus($percent)
    {
        return $this->map(function ($item) use ($percent) {
            $item["bonus"] += $item["amount"] * $percent / 100.0;
            return $item;
        });
    }

    public function getDoneBonusSum()
    {
        return $this
            ->filter_state(self::DONE)
            ->calculate_sum("bonus");
    }

    public function sample_run()
    {
        return $this
            ->filter_date("2017-07-21", "2017-07-23")
            ->map_bonus(0.5)
            ->getDoneBonusSum();
    }
}

/*
 * Sample data
 */
$orders = array(
    ["order_id" => 1, "amount" => 1000, "bonus" => null, "date" => "2017-07-20T18:17:03+08:00", "state" => Order::PENDING],
    ["order_id" => 2, "amount" => 3000, "bonus" => null, "date" => "2017-07-21T21:38:24+08:00", "state" => Order::NEW],
    ["order_id" => 3, "amount" => 2500, "bonus" => null, "date" => "2017-07-22T09:41:51+08:00", "state" => Order::DONE],
    ["order_id" => 4, "amount" => 1800, "bonus" => null, "date" => "2017-07-23T13:24:31+08:00", "state" => Order::PENDING],
    ["order_id" => 5, "amount" => 1200, "bonus" => null, "date" => "2017-07-24T07:55:47+08:00", "state" => Order::DONE],
);


/*
 * Test procedure
 */
print_r(
    [
        "result" => (new Order($orders))->sample_run()->to_rows()
    ]
);

?>

From SQL jams to pipeline style php

From SQL jams to pipeline style php

Pipeline style

初步實作

進一步改善: PipelineArray

進一步封裝: Order class

附註

CATALOG

FEATURED TAGS