Flume 流向 mysql

Question

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

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 数据库的类型。所以我的问题是如何声明 jdbcSink.type?

Answer 1

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

随时从中获得灵感：https://github.com/telefonicaid/fiware-cygnus/blob/master/cygnus-ngsi/src/main/java/com/telefonica/iot/cygnus/sinks/NGSIMySQLSink.java

它为我们所有的接收器扩展了另一个自定义基础 class：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)

分别

Flume 流向 mysql

Flume stream to mysql

mysql

flume

flume-ng