kafka_open()
Description:
Connect to Kafka server.
Syntax:
kafka_open(filename or fileObject; topic,...; partitionSize)
Note:
The external library function connects to Kafka server, and stores attribute parameter properties in the configuration file with extension .properties. It supports single-machine pattern and cluster pattern; it is recommended that the machine holding the leader is used when you are trying to connect to the server through a cluster.
Parameters:
|
topic |
Query one or more specified topics |
|
filename |
The attribute parameter file with the extension .properties, which contains the send key, value encoding and message decoding; all of them should correspond to each other one by one |
|
fileObject |
A file object |
|
partitionSize |
The number of zones on a topic, which can be absent and is valid only in the context of a cluster |
Options:
|
@c |
For the use of clustering and related operations |
Return value:
The consumer object
Example:
|
|
A |
|
|
1 |
=kafka_open("D://kafka.properties";"topic-test") |
Connect topic-test’s configuration file kafka.properties to Kafka server using a topic |
|
2 |
=kafka_open@c(file("D://kafka.properties");"topic-test";3) |
Connect topic-test’s configuration file kafka.properties to Kafka server using a file object |
|
3 |
=kafka_close(A1) |
|
Below is the content of D://kafka.properties:
##produce
bootstrap.servers=192.168.0.1:9092
producer.type=sync
request.required.acks=1
serializer.class=kafka.serializer.DefaultEncoder
key.serializer=org.apache.kafka.common.serialization.StringSerializer
value.serializer=org.apache.kafka.common.serialization.StringSerializer
##consume
group.id=test
zookeeper.session.timeout.ms=1000
zookeeper.sync.time.ms=200
auto.commit.interval.ms=500
auto.offset.reset=earliest
key.deserializer=org.apache.kafka.common.serialization.StringDeserializer
value.deserializer=org.apache.kafka.common.serialization.StringDeserializer
Explanations of user-defined coding and
decoding:
Put the jar containing encoding and decoding information in
exlib\KafkaCil directory, as shown below:
In raq-kafa-cil-2.11.jar, the encoding file is EncodeingSequence.class,
the decoding file is DecodeingSequence.class, and their corresponding
record object is Sequence.
Below are configurations of user-defined encoding and decoding in .properties
file:
key.serializer=org.apache.kafka.common.serialization.StringSerializer
value.serializer=com.raqsoft.lib.kafka.EncodeingSequence
key.deserializer=org.apache.kafka.common.serialization.StringDeserializer value.deserializer=com.raqsoft.lib.kafka.DecodeingSequence