Flume怎样自界说Sink数据至MySQL

2024-11-3 21:34| 发布者: 76a9| 查看: 98| 评论: 0

摘要: 目次Flume自界说Sink数据至MySQL一、先容二、需求三、编写MySink四、编写Flume脚本五、测试总结Flume自界说Sink数据至MySQL 一、先容 Sink不断地轮询Channel中的事件且批量地移除它们，并将这些事件批量写入到存储或

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的接口：

[code]https://flume.apache.org/FlumeDeveloperGuide.html#sink根据官方说明自界说MySink需要继续AbstractSink类并实现Configurable接口。[/code]

实现相应方法：

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

利用场景：

读取Channel数据写入MySQL或者其他文件体系。

二、需求

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

三、编写MySink

[code]package com.flume.flume; import org.apache.flume.*; import org.apache.flume.conf.Configurable; import org.apache.flume.sink.AbstractSink; import java.sql.Connection; import java.sql.DriverManager; import java.sql.PreparedStatement; import java.sql.SQLException; public class MySink extends AbstractSink implements Configurable { private String msgPrefix; /** * 用来生存数据,不断调用次方法 * @return * @throws EventDeliveryException */ @Override public Status process() throws EventDeliveryException { //获取sink对应的channnel Channel channel = getChannel(); Connection connection = null; PreparedStatement statement = null; //获取事件对象 Transaction transaction = channel.getTransaction(); try{ //开启事件 transaction.begin(); //从channel中获取数据 Event event = channel.take(); //切割数据 String data = new String(event.getBody()); String[] arr = data.split(","); String id = arr[0]; String name = arr[1]; int age = Integer.parseInt(arr[2]); //生存到mysql //1、获取connect connection = DriverManager.getConnection("jdbc:mysql://hadoop102:3306/test?useSSL=false","root","123321"); statement = connection.prepareStatement("insert into test values(?,?,?)"); saveToMysql(id,name,age,connection,statement); //模仿数据生存 //System.out.println(msgPrefix+":"+new String(take.getBody())); //提交事件 transaction.commit(); return Status.READY; }catch (Exception e){ transaction.rollback(); }finally { //关闭事件 transaction.close(); if(statement!=null) //5、关闭 { try { statement.close(); } catch (SQLException e) { e.printStackTrace(); } } if(connection!=null) { try { connection.close(); } catch (SQLException e) { e.printStackTrace(); } } } return Status.BACKOFF; } public void saveToMysql(String id,String name,int age,Connection connection,PreparedStatement statement) throws SQLException { //2、获取statement对象 //sql注入【 select * from table where name='zhangsan' or 1=1】 //connection.createStatement(); //3、赋值 statement.setString(1,id); statement.setString(2,name); statement.setInt(3,age); System.out.println(id+","+name+","+age); //4、生存 statement.executeUpdate(); } /** * 获取sink的配置属性 * @param context */ @Override public void configure(Context context) { msgPrefix = context.getString("msg.prefix"); } } [/code]

四、编写Flume脚本

[code]#界说agent a1.sources = r1 a1.channels = c1 a1.sinks = k1 #界说source a1.sources.r1.type = netcat a1.sources.r1.bind = 0.0.0.0 a1.sources.r1.port = 9999 #界说channel a1.channels.c1.type = memory a1.channels.c1.capacity = 1000 a1.channels.c1.transactionCapacity = 1000 #界说sink a1.sinks.k1.type = com.atguigu.flume.MySink a1.sinks.k1.msg.prefix = message #界说source、channel、sink之间的绑定关系 a1.sources.r1.channels = c1 a1.sinks.k1.channel = c1[/code]

五、测试

1.启动flume

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

2.启动nc端口

[code][hadoop@hadoop102 ~]$ nc hadoop102 9999 1,ttt,8 OK[/code]

3.客户端输出