Flume如何自定义Sink数据至MySQL

青年管理者

Flume自定义Sink数据至MySQL


一、介绍

Sink不断地轮询Channel中的事件且批量地移除它们，并将这些事件批量写入到存储或索引系统、或者被发送到另一个Flume Agent。
Sink是完全事务性的。在从Channel批量删除数据之前，每个Sink用Channel启动一个事务。批量事件一旦成功写出到存储系统或下一个Flume Agent，Sink就利用Channel提交事务。事务一旦被提交，该Channel从自己的内部缓冲区删除事件。
Sink组件目的地包括hdfs、logger、avro、thrift、ipc、file、null、HBase、solr、自定义。官方提供的Sink类型已经很多，但是有时候并不能满足实际开发当中的需求，此时我们就需要根据实际需求自定义某些Sink。
官方也提供了自定义sink的接口：

1. https://flume.apache.org/FlumeDeveloperGuide.html#sink根据官方说明自定义MySink需要继承AbstractSink类并实现Configurable接口。

复制代码

实现相应方法：

• configure(Context context)//初始化context（读取配置文件内容）
  
• process()//从Channel读取获取数据（event），这个方法将被循环调用。
  

使用场景：
读取Channel数据写入MySQL或者其他文件系统。

二、需求

使用flume接收(id,name,string)数据，并在Sink端给每条数据进行切分，编写JDBC驱动将数据保存到MySQL数据库。

三、编写MySink

1. package com.flume.flume;
   
2. 
   
3. import org.apache.flume.*;
   
4. import org.apache.flume.conf.Configurable;
   
5. import org.apache.flume.sink.AbstractSink;
   
6. 
   
7. import java.sql.Connection;
   
8. import java.sql.DriverManager;
   
9. import java.sql.PreparedStatement;
   
10. import java.sql.SQLException;
    
11. 
    
12. public class MySink extends AbstractSink implements Configurable {
    
13. 
    
14.     private String msgPrefix;
    
15. 
    
16.     /**
    
17.      * 用来保存数据,不断调用次方法
    
18.      * @return
    
19.      * @throws EventDeliveryException
    
20.      */
    
21.     @Override
    
22.     public Status process() throws EventDeliveryException {
    
23.         //获取sink对应的channnel
    
24.         Channel channel = getChannel();
    
25.         Connection connection = null;
    
26.         PreparedStatement statement = null;
    
27.         //获取事务对象
    
28.         Transaction transaction = channel.getTransaction();
    
29.         try{
    
30.             //开启事务
    
31.             transaction.begin();
    
32.             //从channel中获取数据
    
33.             Event event = channel.take();
    
34. 
    
35.             //切割数据
    
36.             String data = new String(event.getBody());
    
37.             String[] arr = data.split(",");
    
38.             String id = arr[0];
    
39.             String name = arr[1];
    
40.             int age = Integer.parseInt(arr[2]);
    
41. 
    
42.             //保存到mysql
    
43.             //1、获取connect
    
44.             connection = DriverManager.getConnection("jdbc:mysql://hadoop102:3306/test?useSSL=false","root","123321");
    
45.             statement = connection.prepareStatement("insert into test values(?,?,?)");
    
46.             saveToMysql(id,name,age,connection,statement);
    
47.             //模拟数据保存
    
48.             //System.out.println(msgPrefix+":"+new String(take.getBody()));
    
49.             //提交事务
    
50.             transaction.commit();
    
51. 
    
52.             return Status.READY;
    
53.         }catch (Exception e){
    
54.             transaction.rollback();
    
55.         }finally {
    
56.             //关闭事务
    
57.             transaction.close();
    
58.             if(statement!=null)
    
59.             //5、关闭
    
60.             {
    
61.                 try {
    
62.                     statement.close();
    
63.                 } catch (SQLException e) {
    
64.                     e.printStackTrace();
    
65.                 }
    
66.             }
    
67.             if(connection!=null) {
    
68.                 try {
    
69.                     connection.close();
    
70.                 } catch (SQLException e) {
    
71.                     e.printStackTrace();
    
72.                 }
    
73.             }
    
74.         }
    
75. 
    
76.         return Status.BACKOFF;
    
77.     }
    
78. 
    
79.     public void saveToMysql(String id,String name,int age,Connection connection,PreparedStatement statement) throws SQLException {
    
80. 
    
81.         //2、获取statement对象
    
82.         //sql注入 【 select * from table where name='zhangsan' or 1=1】
    
83.         //connection.createStatement();
    
84. 
    
85.         //3、赋值
    
86.         statement.setString(1,id);
    
87.         statement.setString(2,name);
    
88.         statement.setInt(3,age);
    
89.         System.out.println(id+","+name+","+age);
    
90.         //4、保存
    
91.         statement.executeUpdate();
    
92. 
    
93. 
    
94.     }
    
95.     /**
    
96.      * 获取sink的配置属性
    
97.      * @param context
    
98.      */
    
99.     @Override
    
100.     public void configure(Context context) {
     
101. 
     
102.         msgPrefix = context.getString("msg.prefix");
     
103. 
     
104.     }
     
105. }

复制代码

四、编写Flume脚本

1. #定义agent
   
2. a1.sources = r1
   
3. a1.channels = c1
   
4. a1.sinks = k1
   
5. 
   
6. #定义source
   
7. a1.sources.r1.type = netcat
   
8. a1.sources.r1.bind = 0.0.0.0
   
9. a1.sources.r1.port = 9999
   
10. 
    
11. #定义channel
    
12. a1.channels.c1.type = memory
    
13. a1.channels.c1.capacity = 1000
    
14. a1.channels.c1.transactionCapacity = 1000
    
15. 
    
16. #定义sink
    
17. a1.sinks.k1.type = com.atguigu.flume.MySink
    
18. a1.sinks.k1.msg.prefix = message
    
19. 
    
20. #定义source、channel、sink之间的绑定关系
    
21. a1.sources.r1.channels = c1
    
22. a1.sinks.k1.channel = c1

复制代码

五、测试

1.启动flume

1. [hadoop@hadoop102 ~]$ cd /opt/module/flume/
   
2. [hadoop@hadoop102 flume]$ bin/flume-ng agent -c conf/ -n a1 -f job/mysik.config -Dflume.root.logger=INFO,console

复制代码

2.启动nc端口

1. [hadoop@hadoop102 ~]$ nc hadoop102 9999
   
2. 1,ttt,8
   
3. OK

复制代码

3.客户端输出

4.查看MySQL数据库

总结

以上为个人经验，希望能给大家一个参考，也希望大家多多支持脚本之家。

来源:https://www.jb51.net/database/328292d4v.htm
免责声明：由于采集信息均来自互联网，如果侵犯了您的权益，请联系我们【E-Mail:cb@itdo.tech】 我们会及时删除侵权内容，谢谢合作！