...
Code Block |
---|
language | javascript |
---|
theme | Eclipse |
---|
title | File Transport Handler |
---|
linenumbers | true |
---|
|
#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'],
['metadata.broker.list', 'localhost:9092']
]
},
input
) |
PQL for reading
Important: There is no GenericPull-Version of the transport handler. So alway us GenericPush!
Code Block |
---|
language | javascript |
---|
theme | Eclipse |
---|
title | File Transport Handler |
---|
linenumbers | true |
---|
|
#PARSER PQL
#ADDQUERY
input := ACCESS({
source='test',
wrapper='GenericPush',
transport='Kafka',
protocol='SimpleCSV',
datahandler='Tuple',
options=[
['topic', 'test'],
['messagetype', 'string'],
['bootstrap.servers', 'localhost:9092']
],
schema=[
['text', 'String']
]
}
) |