在 Flink 中，可以使用窗口函數（Window Function）對數據進行聚合。以下是一個簡單的示例，演示了如何在 Flink 中使用 PHP 進行數據聚合：

1. 首先，確保已經安裝了 Flink PHP 擴展。可以通過以下命令安裝：

pecl install flink-php

2. 創建一個 PHP 文件，例如 aggregate.php，并編寫以下代碼：

<?php

require_once 'vendor/autoload.php';

use Flink\Common\Type\TypeInformation;
use Flink\Core\Environment;
use Flink\Core\Window;
use Flink\EventTime\EventTimeAttribute;
use Flink\Flink;
use Flink\Sql\SqlFunction;
use Flink\Table\Planner\Planner;
use Flink\Table\Table;
use Flink\Table\TableDescriptor;
use Flink\Table\Types;
use Flink\Table\WindowManager;
use Flink\Table\Windowing;

// 創建 Flink 環境
$env = Environment::getExecutionEnvironment();

// 創建一個表描述符，定義輸入和輸出列
$tableDescriptor = TableDescriptor
    ->forTable("my_table")
    ->withSchema(Types::createSchema(
        Types::field("id", Types::INT())
            ->withName("id")
            ->withType(Types::INT())
    ))
    ->withRowType(Types::createRowType(
        Types::field("value", Types::DOUBLE())
            ->withName("value")
            ->withType(Types::DOUBLE())
    ))
    ->withPrimaryKey("id");

// 創建一個Planner
$planner = Planner::create($env);

// 注冊表
$table = $planner->createTable($tableDescriptor);

// 定義一個窗口函數
$windowFunction = SqlFunction::create("SUM", TypeInformation::of(Types::DOUBLE()), "value");

// 定義一個窗口
$window = Window::create(Windowing::createTumblingEventTimeWindows(Time::minutes(5)));

// 注冊窗口函數
Planner::getPlanner()->registerFunction("sum", $windowFunction);

// 執行聚合操作
$table->groupBy($window, "id")
    ->select("id, sum(value) as total_value")
    ->execute()
    ->print();

// 啟動 Flink 作業
Flink::run($env, "aggregate.php");

在這個示例中，我們創建了一個名為 my_table 的表，并使用窗口函數對每 5 分鐘的數據進行求和。最后，我們打印出聚合結果。

要運行此示例，請確保已經安裝了 Flink，并將 aggregate.php 放在 Flink 的 PHP 可執行文件所在的目錄中。然后，可以通過以下命令運行 Flink 作業：

./bin/flink run -c com.example.Aggregate aggregate.php

請注意，這個示例僅用于演示如何在 Flink 中使用 PHP 進行數據聚合。在實際應用中，可能需要根據具體需求進行調整。