Flume 流到 mysql答案

【问题标题】：Flume stream to mysqlFlume 流到 mysql
【发布时间】：2017-02-20 02:12:59
【问题描述】：

我一直在尝试使用 APACHE KAFKA 和 FLUME 将数据流式传输到 MySQL 数据库中。（这是我的水槽配置文件）

agent.sources=kafkaSrc
agent.channels=channel1
agent.sinks=jdbcSink

agent.channels.channel1.type=org.apache.flume.channel.kafka.KafkaChannel
agent.channels.channel1.brokerList=localhost:9092
agent.channels.channel1.topic=kafkachannel
agent.channels.channel1.zookeeperConnect=localhost:2181
agent.channels.channel1.capacity=10000
agent.channels.channel1.transactionCapacity=1000


agent.sources.kafkaSrc.type = org.apache.flume.source.kafka.KafkaSource
agent.sources.kafkaSrc.channels = channel1
agent.sources.kafkaSrc.zookeeperConnect = localhost:2181
agent.sources.kafkaSrc.topic = kafka-mysql

***agent.sinks.jdbcSink.type = How to declare this?***
agent.sinks.jdbcSink.connectionString = jdbc:mysql://1.1.1.1:3306/test
agent.sinks.jdbcSink.username=user
agent.sinks.jdbcSink.password=password
agent.sinks.jdbcSink.batchSize = 10
agent.sinks.jdbcSink.channel =channel1
agent.sinks.jdbcSink.sqlDialect=MYSQL
agent.sinks.jdbcSink.driver=com.mysql.jdbc.Driver
agent.sinks.jdbcSink.sql=(${body:varchar})

我知道如何将数据流式传输到 hadoop 或 hbase（记录器类型或 hdfs 类型），但是找不到流式传输到 mysql DB 的类型。所以我的问题是如何声明 jdbcSink.type？

【问题讨论】：

flume 中没有 JDBC 接收器。您不能使用 Flume 将数据流式传输到 MySQL。
@franklinsijo 感谢您的回复。那么有没有办法将数据从 Kafka 拉到 RDBMS 中呢？我愿意接受任何建议。
用Kafka JDBC Sink不可能吗？

标签： mysql flume flume-ng

【解决方案1】：

您总是可以为 MySQL 创建一个自定义接收器。这就是我们在 FIWARE 使用 Cygnus 工具所做的。

随时从中获得灵感：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 接口。这意味着至少要覆盖以下方法：

public Status process() throws EventDeliveryException

和：

public void configure(Context context)

分别。

【讨论】：

非常感谢。