Versions Compared

Key

  • This line was added.
  • This line was removed.
  • Formatting was changed.

...

  • topic: String defining the kafka topic to read/write
  • messagetype: either 'string' or 'bytearray'; defines the type of values written to/read from Kafka

Mandatory for writing;

  • metadata.broker.list: comma separated list of Kafka brokers (host:port)

Mandatory for reading;

  • bootstrap.servers: comma separated list of Kafka brokers (host:port)

Optional:

  • All other properties possible for Kafka Producers and Consumers respectively (see Kafka documentation for possible properties)

Please note that the parameters are directly forwarded to kafka, so, please also check the Kafka documentation.

Example

PQL for writing

Code Block
languagejavascript
themeEclipse
titleFile Transport Handler
linenumberstrue
#PARSER PQL
#RUNQUERY
input = ACCESS({
            source='Input',
            wrapper='GenericPull',
            transport='file',
            protocol='SimpleCSV',
            datahandler='Tuple',
            options=[
              ['filename', '${PROJECTPATH}/input.csv'],
              ['csv.delimiter', ','],
              ['csv.trim', 'true']
            ],
            schema=[
              ['text', 'String']
            ]                                                                                                                                                                                                                                                        
          }                                                                                                                                                                                                                                  
        )
        
output = SENDER({
              protocol = 'SimpleCSV',
              transport = 'Kafka',
              sink = 'Test',
              wrapper = 'GenericPush',
              options = [
                ['topic', 'test'], 
                ['messagetype', 'string'],
                ['metadatabootstrap.broker.listservers', 'localhost:9092']
              ]
            },
            input
          )

...

Important: There is no GenericPull -Version version of the transport handler. So alway us GenericPush!

...