Flume 之自定义Sink

1、简介

        前文我们介绍了 Flume 如何自定义 Source， 并进行案例演示，本文将接着前文，自定义Sink，在这篇文章中，将使用自定义 Source 和 自定义的 Sink 实现数据传输，让大家快速掌握Flume这门技术。

2、自定义Source

        自定义Source参考前文：https://blog.csdn.net/zwl2220943286/article/details/135633120

3、自定义Sink

        本文将Sink定义为mysql。

3.1、引入依赖

<dependency><groupId>org.apache.flume</groupId><artifactId>flume-ng-core</artifactId><version>1.11.0</version>
</dependency><dependency><groupId>mysql</groupId><artifactId>mysql-connector-java</artifactId><version>8.0.33</version>
</dependency>

3.2、自定义Sink

3.2.1、Sink代码

import com.weilong.flumeselfdefinition.util.MysqlConfig;
import org.apache.flume.*;
import org.apache.flume.conf.Configurable;
import org.apache.flume.sink.AbstractSink;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;public class MySink extends AbstractSink implements Configurable {private final static Logger log = LoggerFactory.getLogger(MySink.class);private String url;private String username;private String password;@Overridepublic Status process() throws EventDeliveryException {Status status = null;Channel channel = getChannel();// channel 支持事务Transaction thx = channel.getTransaction();thx.begin();try {Event event = channel.take();String name = new String(event.getBody());int i = MysqlConfig.insertData(this.url, this.username, this.password, name);if (i > 0){log.info("==插入数据库成功==");}thx.commit();status = Status.READY;} catch (Exception ex){ex.printStackTrace();}return status;}@Overridepublic void configure(Context context) {String url = context.getString("url");String username = context.getString("username");String password = context.getString("password");this.url = url;this.username = username;this.password = password;}
}

 3.2.2、数据库连接配置：

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;public class MysqlConfig {private MysqlConfig(){}static {try {Class.forName("com.mysql.cj.jdbc.Driver");}catch (Exception ex){ex.printStackTrace();}}public static Connection getConnection(String url, String username, String password) throws SQLException {Connection connection = DriverManager.getConnection(url, username, password);return connection;}public static int insertData(String url, String username,String password, String name){Connection connection = null;try{connection = getConnection(url, username, password);PreparedStatement preparedStatement = connection.prepareStatement("insert into test(`name`) values( '" + name + "')");boolean res = preparedStatement.execute();if (res){return 1;}return 0;}catch (Exception ex){ex.printStackTrace();}finally {if (connection != null){try {connection.close();}catch (Exception ex){ex.printStackTrace();}}}return 0;}
}

 3.3、Flume 配置文件

        vim flume-self-source-sink.conf

a1.sources = r1
a1.channels = c1
a1.sinks=k1
# source
a1.sources.r1.type = com.weilong.flumeselfdefinition.MySource 
# 自定义 Source 的全限定类名
a1.sources.r1.path = http://192.168.30.3:8088/hello 
# 自定义参数
# channel
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
# 自定义Sink
a1.sinks.k1.type = com.weilong.flumeselfdefinition.MySink
a1.sinks.k1.url = jdbc:mysql://192.168.30.3:3306/test?useUnicode=true&characterEncoding=utf8&serverTimezone=UTC
a1.sinks.k1.username = root
a1.sinks.k1.password = 146815
# bind
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

4、将jar包放入lib目录 

 4.1、将自定义jar包放入lib目录

 

4.2、将数据库驱动jar包放入lib目录

        驱动jar包下载地址：https://mvnrepository.com/artifact/mysql/mysql-connector-java

注：mysql 驱动jar包不放进lib，会出现驱动类找不到。 

5、启动 Flume

bin/flume-ng agent -c conf/ -n a1 -f testconf/flume-self-source-sink.conf -Dflume.root.logger=INFO,console

注：启动Flume 之前，自定义 web 服务也要启动。

 6、结果

成功保存进数据库。

7、总结 

         本文结合前文完成 Flume 的 Source 和 Sink 的自定义，帮助大家能够完成各种场景下的Flume的使用。关于更高级Flume的知识，关注下面公众号。

        本人是一个从小白自学计算机技术，对运维、后端、各种中间件技术、大数据等有一定的学习心得，想获取自学总结资料（pdf版本）或者希望共同学习，关注微信公众号：it自学社团。后台回复相应技术名称/技术点即可获得。（本人学习宗旨：学会了就要免费分享）