php ksql

Ytake KsqlClient [ytake / php-ksql]

Apache kafka / Confluent KSQL REST ไคลเอ็นต์สำหรับ php.ini

KSQL คืออะไร

KSQL เป็นเครื่องมือสตรีมมิ่ง SQL สำหรับ Apache Kafka

KSQL คืออะไร?

ติดตั้ง

จำเป็น >= PHP 7.1

$ composer require ytake/php-ksql

การใช้งาน

ขอค่าที่ตั้งไว้ล่วงหน้า

ระดับ
YtakeKsqlClientQueryCommandStatus
YtakeKsqlClientQueryStatus
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 ();