php ksql

YtakeKsqlClient [ytake/php-ksql]

Apache kafka / Cliente REST KSQL Confluent para php

O que é KSQL

KSQL é o mecanismo SQL de streaming para Apache Kafka.

O que é KSQL?

Instalar

obrigatório >= PHP 7.1

$ composer require ytake/php-ksql

Uso

Solicitar predefinição

aula
YtakeKsqlClientQueryCommandStatus
YtakeKsqlClientQueryStatus
YtakeKsqlClientQueryServerInfo
YtakeKsqlClientQueryKsql
YtakeKsqlClientQueryStream (para fluxo)

Referência de sintaxe

Obter status do comando

 <?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 ();

Obter status

 <?php

use Ytake  KsqlClient  RestClient ;
use Ytake  KsqlClient  Query  Status ;

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

Obtenha informações do servidor KSQL

 <?php

use Ytake  KsqlClient  RestClient ;
use Ytake  KsqlClient  Query  ServerInfo ;

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

Consultar 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 ();

Cliente para resposta de fluxo

 <?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 ();