php ksql

YtakeKsqlClient [ytake/php-ksql]

Apache kafka / 適用於 php 的 Confluence KSQL REST 用戶端

什麼是 KSQL

KSQL 是 Apache Kafka 的串流 SQL 引擎。

什麼是 KSQL？

安裝

需要 >= PHP 7.1

$ composer require ytake/php-ksql

用法

請求預設

班級
YtakeKsqlClientQueryCommandStatus
YtakeKsqlClient查詢狀態
YtakeKsqlClientQueryServerInfo
YtakeKsqlClientQueryKsql
YtakeKsqlClientQueryStream（對於流）

語法參考

取得命令狀態

 <?php

use Ytake  KsqlClient  RestClient ;
use Ytake  KsqlClient  Query  CommandStatus ;
use Ytake  KsqlClient  Computation  CommandId ;

$ client = new RestClient (
    " http://localhost:8088 "
);
$ result = $ client -> requestQuery (
    new CommandStatus (CommandId:: fromString ( ' stream/MESSAGE_STREAM/create ' ))
)-> result ();

獲取狀態

 <?php

use Ytake  KsqlClient  RestClient ;
use Ytake  KsqlClient  Query  Status ;

$ client = new RestClient (
    " http://localhost:8088 "
);
$ result = $ client -> requestQuery ( new Status ())-> result ();

取得 KSQL 伺服器信息

 <?php

use Ytake  KsqlClient  RestClient ;
use Ytake  KsqlClient  Query  ServerInfo ;

$ client = new RestClient (
    " http://localhost:8088 "
);
$ result = $ client -> requestQuery ( new ServerInfo ())-> result ();

查詢KSQL

 <?php

use Ytake  KsqlClient  RestClient ;
use Ytake  KsqlClient  Query  Ksql ;

$ client = new RestClient (
    " http://localhost:8088 "
);
$ result = $ client -> requestQuery (
    new Ksql ( ' DESCRIBE users_original; ' )
)-> result ();

串流響應客戶端

 <?php

use Ytake  KsqlClient  StreamClient ;
use Ytake  KsqlClient  Query  Stream ;
use Ytake  KsqlClient  StreamConsumable ;
use Ytake  KsqlClient  Entity  StreamedRow ;

$ client = new StreamClient (
    " http://localhost:8088 "
);
$ result = $ client -> requestQuery (
    new Stream (
        ' SELECT * FROM testing ' ,
        new class () implements StreamConsumable {
            public function __invoke ( StreamedRow $ row ) 
            {
                // stream response consumer
            }
        }    
    )
)-> result ();