Kafka是一种分布式的信息流平台，它的客户端分为生产者、消费者两种，生产者提交数据，消费者读取数据。Kafka中传递的事件消息没有固定格式，消费者、生产者自由约定一些个性化格式，比如是JSON，XML等结构的一个大字符串，SPL能很好的解析这些数据，后续也能做一些复杂计算。

创建/关闭Kafka连接

使用方式类似关系数据库的JDBC连接，SPL也用成对的“创建/关闭”方式连接Kafka。

kafka_open(propertiesFile; topic1, topic2, ...)， propertiesFile是连接参数，topic是Kafka主题，可以多个，从多主题中同时读写数据。

propertiesFile示例如下，更多参数可以参考Kafka官方文档

kafka_close(kafkaConn)，kafkaConn是要关闭的Kafka连接。

示例：A1创建连接，中间做一些读写及计算操作后，A3关闭A1创建的连接

生产者发送消息

kafka_send(kafkaConn, [key], value) ，kafkaConn是Kafka连接，key是消息的健，可以没有，value是消息内容。

示例：

A2、A3是两条待提交的JSON格式事件，A4,A5用kafka_send()函数提交到Kafka平台。

消费者读取消息

kafka_poll(kafkaConn) ，kafkaConn是Kafka连接，默认读取所有未读取的数据：

A2读到的value是一个JSON字符串

A3用json()函数解析JSON字符串，用derive()、new()函数整理数据，形成规整的数据表：

如果数据量太大，可以加@c选项，这样就可以用游标方式分批加载数据。

执行后，A2中看不到数据，它只是一个游标对象。A3取第1条数据：

A4取之后的10行数据（实际上一共只有2条数据，所以这次只取到最后1条）：