...
- 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 | ||||||||
|---|---|---|---|---|---|---|---|---|
| ||||||||
#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
) |
...