我一直在尝试使用APACHE KAFKA和FLUME将数据流式传输到MySQL数据库。 (这是我的flume配置文件)
agent.sources = kafkaSrcagent.channels =通道1agent.sinks = jdbcSink
代理….
您总是可以为MySQL创建自定义接收器。这就是我们在FIWARE上所做的 天鹅 工具。
随意从中获取灵感: https://github.com/telefonicaid/fiware-cygnus/blob/master/cygnus-ngsi/src/main/java/com/telefonica/iot/cygnus/sinks/NGSIMySQLSink.java
它扩展了我们所有接收器的另一个自定义基类: https://github.com/telefonicaid/fiware-cygnus/blob/master/cygnus-ngsi/src/main/java/com/telefonica/iot/cygnus/sinks/NGSISink.java
基本上,你必须扩展 AbstractSink 并实施 Configurable 接口。这意味着至少要覆盖以下方法:
AbstractSink
Configurable
public Status process() throws EventDeliveryException
和:
public void configure(Context context)
分别。