如何在PHP开发中使用Apache Beam实现批处理和流处理的统一接口和架构

sudo apt-get install -y php-pear curl php7.x-dev libcurl4-openssl-dev
sudo pecl install grpc protobuf

composer require apache/beam-php-sdk

use ApacheBeamCreate;
use ApacheBeamExamplesCompleteJSONToMySQLJSONToMySQLMySQLConfiguration;
use ApacheBeamPipelineBuilder;

class JsonToMySqlPipeline
{
    private $pipelineBuilder;
    private $input;
    private $output;

    public function __construct($input, $output)
    {
        $this->pipelineBuilder = new PipelineBuilder([
            'appName' => 'json-to-mysql-pipeline'
        ]);
        $this->input = $input;
        $this->output = $output;
    }

    public function build()
    {
        $this->pipelineBuilder
            ->apply(Create::fromArray([[$this->input]]))
            ->apply(
                'Transform JSON to Associative Array',
                MapElements::into(
                    DataTypes::ARRAY(
                        DataTypes::STRING(), DataTypes::STRING()
                    )
                )->via(
                    function ($json) {
                        $data = json_decode($json, true);
                        return [
                            'name' => $data['name'],
                            'age' => $data['age']
                        ];
                    }
                )
            )
            ->apply(
                'Write to MySQL',
                new WriteToMySQL(
                    $this->output,
                    new MySQLConfiguration(
                        $host = 'localhost',
                        $port = '3306',
                        $user = 'root',
                        $password = '',
                        $database = 'beam',
                        $table = 'users'
                    )
                )
            );
    }

    public function run()
    {
        $this->pipelineBuilder->run();
    }
}

$input = 'data/users.json';
$output = 'mysql';

$pipeline = new JsonToMySqlPipeline($input, $output);
$pipeline->build();
$pipeline->run();