flume之自定义sink组件

阳光穿透心脏的1/2处 2021-06-24 15:57 683阅读 0赞

flume内部提供了很多种sink，如：logger、file_roll、avro、hdfs、kafak、es等，方便直接将event数据对接到本地磁盘、或者其他第三方存储中。有的时候，我们需要自定义source，来完成特殊需求。本文介绍如何开发自定义sink，来实现将event数据存储到Mysql。

1、pom.xml

  1. <?xml version="1.0"?>
  2. <project xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd" xmlns="http://maven.apache.org/POM/4.0.0"
  3.     xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
  4.   <modelVersion>4.0.0</modelVersion>
  5.   <parent>
  6.     <groupId>com.abc</groupId>
  7.     <artifactId>ttbrain-log</artifactId>
  8.     <version>0.0.1-SNAPSHOT</version>
  9.   </parent>
  11.   <groupId>com.abc</groupId>
  12.   <artifactId>ttbrain-log-flume</artifactId>
  13.   <version>0.0.1-SNAPSHOT</version>
  14.   <name>ttbrain-log-flume</name>
  16.   <properties>
  17.     <version.flume>1.7.0</version.flume>
  18.   </properties>
  21.   <dependencies>
  22.     <dependency>
  23.       <groupId>junit</groupId>
  24.       <artifactId>junit</artifactId>
  25.       <scope>test</scope>
  26.     </dependency>
  27.     <dependency>
  28.         <groupId>org.slf4j</groupId>
  29.         <artifactId>slf4j-log4j12</artifactId>
  30.     </dependency>
  32.     <!-- flume -->
  33.     <dependency>
  34.        <groupId>org.apache.flume</groupId>
  35.        <artifactId>flume-ng-core</artifactId>
  36.        <version>${version.flume}</version>
  37.     </dependency>
  38.     <dependency>
  39.        <groupId>org.apache.flume</groupId>
  40.        <artifactId>flume-ng-configuration</artifactId>
  41.        <version>${version.flume}</version>
  42.     </dependency>
  44.     <!-- mysql -->
  45.     <dependency>
  46.         <groupId>c3p0</groupId>
  47.         <artifactId>c3p0</artifactId>
  48.     </dependency>
  49.     <dependency>
  50.         <groupId>com.alibaba</groupId>
  51.         <artifactId>druid</artifactId>
  52.     </dependency>
  53.     <dependency>
  54.         <groupId>mysql</groupId>
  55.         <artifactId>mysql-connector-java</artifactId>
  56.     </dependency>
  57.   </dependencies>
  59.   <profiles>
  60.         <profile>
  61.             <id>dev</id>
  62.             <properties>
  63.                 <profile.env.name>dev</profile.env.name>
  64.             </properties>
  65.             <activation>
  66.                 <activeByDefault>true</activeByDefault>
  67.             </activation>
  68.         </profile>
  69.         <profile>
  70.             <id>test</id>
  71.             <properties>
  72.                 <profile.env.name>test</profile.env.name>
  73.             </properties>
  74.         </profile>
  75.         <profile>
  76.             <id>product</id>
  77.             <properties>
  78.                 <profile.env.name>product</profile.env.name>
  79.             </properties>
  80.         </profile>
  81.     </profiles>
  83.     <build>
  84.     <finalName>ttbrain-log-flume-MysqlSink</finalName>
  85.     <filters>
  86.             <filter>${basedir}/filters/filter-${profile.env.name}.properties</filter><!--这里指定filter属性文件的位置-->
  87.     </filters>
  88.     <resources>
  89.         <resource>
  90.             <directory>src/main/resources</directory>
  91.             <filtering>true</filtering><!--这里开启变量替换-->
  92.             <includes>
  93.                 <include>**/*.xml</include>
  94.                 <include>conf/*.properties</include>
  95.                 <include>**/*.properties</include>
  96.                 <include>**/*.json</include>
  97.             </includes>
  98.         </resource>
  99.     </resources>
  100.     <plugins>
  101.             <!-- <plugin>
  102.                 <groupId>org.apache.maven.plugins</groupId>
  103.                 <artifactId>maven-jar-plugin</artifactId>
  104.                 <version>2.4</version>
  105.                 <configuration>
  106.                     <archive>
  107.                         <manifest>
  108.                             <addClasspath>true</addClasspath>
  109.                             <classpathPrefix>lib/</classpathPrefix>
  110.                             <mainClass>com.abc.ttbrain.log.flume.interceptor.MyInterceptor</mainClass>
  111.                         </manifest>
  112.                         <manifestEntries>
  113.                             <Class-Path>conf/</Class-Path>
  114.                         </manifestEntries>
  115.                     </archive>
  116.                     <includes>
  117.                         <include>**/*.class</include>
  118.                     </includes>
  119.                 </configuration>
  120.             </plugin> -->
  121.             <plugin>
  122.                 <groupId>org.apache.maven.plugins</groupId>
  123.                 <artifactId>maven-assembly-plugin</artifactId>
  124.                 <version>2.4</version>
  125.                 <configuration>
  126.                     <descriptorRefs>  
  127.                         <descriptorRef>jar-with-dependencies</descriptorRef>  
  128.                     </descriptorRefs>
  129.                     <archive>
  130.                         <manifest>
  131.                             <mainClass>com.abc.ttbrain.log.flume.sink.MysqlSink</mainClass>
  132.                         </manifest>
  133.                     </archive> 
  134.                 </configuration>
  135.                 <executions>
  136.                     <execution>
  137.                         <id>make-assembly</id>
  138.                         <phase>package</phase>
  139.                         <goals>
  140.                             <goal>single</goal>
  141.                         </goals>
  142.                     </execution>
  143.                 </executions>
  144.             </plugin>
  145.         </plugins>
  146.   </build>
  147. </project>

2、开发自定义sink，继承AbstractSink

  1. package com.abc.ttbrain.log.flume.sink;
  3. import java.sql.Connection;
  4. import java.sql.PreparedStatement;
  5. import java.sql.SQLException;
  6. import java.util.HashSet;
  7. import java.util.List;
  8. import java.util.Set;
  9. import org.apache.commons.lang.StringUtils;
  10. import org.apache.flume.Channel;
  11. import org.apache.flume.Context;
  12. import org.apache.flume.Event;
  13. import org.apache.flume.EventDeliveryException;
  14. import org.apache.flume.Transaction;
  15. import org.apache.flume.conf.Configurable;
  16. import org.apache.flume.sink.AbstractSink;
  17. import org.slf4j.Logger;
  18. import org.slf4j.LoggerFactory;
  19. import com.google.common.collect.Lists;
  20. import com.abc.ttbrain.log.flume.sink.db.DataSourceUtils;
  22. /**
  23.  * MysqlSink 
  24.  * @author kevinliu
  25.  *
  26.  */
  27. public class MysqlSink extends AbstractSink implements Configurable {
  29.     private static final Logger logger = LoggerFactory.getLogger(MysqlSink.class);
  31.     private String tableName = "ttengine_history";
  32.     private int batchSize = 100;
  34.     private Set<String> deviceKeySet = new HashSet<>();
  35.     private  String sql = "";
  37.     public MysqlSink() {
  38.         logger.info("MysqlSink start...");
  39.     }
  41.     @Override
  42.     public void start() {
  43.         super.start();
  44.         deviceKeySet.add("0B466578-F542-4928-A66B-B12FCFA6DA23");
  45.         deviceKeySet.add("CB13721AE4668DEA151F3F72877A2256");
  46.         deviceKeySet.add("033C179D-8987-4D08-AF4F-94070F8632E0");
  47.         deviceKeySet.add("0CD570CE-4B62-45AA-B86A-7828CFE4F195");
  48.         deviceKeySet.add("D9157D45-C17E-4CF9-9C28-EE573F8CAD19");
  49.         deviceKeySet.add("3ACE3422774FEE190304604277E389F9");
  50.         deviceKeySet.add("2C4B1E38-2629-4AF9-BA0E-2DA17C5F3A35");
  51.         deviceKeySet.add("A1AD2FAF-86DD-45DE-B85D-7FA23CCDE4C4");
  53.         logger.info("deviceKey:"+deviceKeySet.toString());
  55.         StringBuilder sb = new StringBuilder();
  56.         sb.append("insert into ").append(tableName)
  57.         .append(" (uid,ppuid,ch_id,f_num,cost,usg,prior,req_id,vers,rg,rh,pg,ph,sg,sh,time_stamp,host,"
  58.                 + "rec_feed_ids,txt,vedio,gallery,p_1,p_2,p_3,p_4,p_5,p_6,p_7,p_8,p_9,p_10,p_11,p_12,p_13,p_14,p_15"
  59.                 + ",p_16,p_17,p_18,p_19,p_20,p_21,p_22,p_23,p_24,p_25,p_26,p_27,p_28,p_29,p_30,p_31,p_32"
  60.                 + ",p_33,p_34,p_35,p_36,p_37,p_38,p_39,p_40,p_41,p_42,p_43,p_44,p_45,p_46,p_47,p_48,p_49,p_50"
  61.                 + ") values")
  62.         .append(" (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,"
  63.                 + "?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?"
  64.                 + ",?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) ");
  66.         sql = sb.toString();
  67.     }
  69.     @Override
  70.     public void stop() {
  71.         super.stop();
  73.         //DataSourceUtils.closeDs();
  74.     }
  76.     @Override
  77.     public Status process() throws EventDeliveryException {
  78.         Status result = Status.READY;
  79.         Transaction transaction = null;
  80.         Event event = null;
  81.         String content = "";
  82.         List<String> actions = Lists.newArrayList();
  84.         PreparedStatement preparedStatement = null;
  85.         Connection conn = null;
  86.         try {
  87.             //db
  88.             conn = DataSourceUtils.getConnection();
  89.             preparedStatement = conn.prepareStatement(sql);
  91.             //flume
  92.             Channel channel = getChannel();
  93.             transaction = channel.getTransaction();
  94.             transaction.begin();
  96.             for (int i = 0; i < batchSize; i++) {
  97.                 event = channel.take();
  98.                 if (event == null) {
  99.                     result = Status.BACKOFF;
  100.                     break;
  101.                 } else {
  102.                     content = new String(event.getBody(),"UTF-8");
  103.                     String[] split = content.split("\t");
  104.                     if (split.length < 37) {
  105.                         continue;
  106.                     }
  107.                     String uid = split[0];
  108.                     boolean contains = deviceKeySet.contains(uid);
  109.                     if (!contains) {
  110.                         continue;
  111.                     }
  112.                     //logger.info(content);
  113.                     actions.add(content);
  114.                 }
  115.             }
  117.             if (actions.size() > 0) {
  118.                 conn.setAutoCommit(false);
  119.                 preparedStatement.clearBatch();
  120.                 //logger.info(actions.get(0));
  121.                 for (String line : actions) {
  122.                     String[] split = line.split("\t");
  123.                     //uid,ppuid,ch_id,f_num,cost,
  124.                     preparedStatement.setString(1, split[0]);//uid
  125.                     preparedStatement.setString(2, split[1]);//ppuid
  126.                     preparedStatement.setString(3, split[2]);//ch_id
  127.                     preparedStatement.setInt(4, Integer.parseInt(split[3]));//f_num
  128.                     preparedStatement.setInt(5, Integer.parseInt(split[4]));//cost
  130.                     //usg,prior,req_id,vers,rg,rh,pg,ph,sg,sh,time_stamp,host,
  131.                     preparedStatement.setInt(6, Integer.parseInt(split[5]));//usg
  132.                     preparedStatement.setString(7, split[6]);//prior
  133.                     preparedStatement.setString(8, split[7]);//req_id
  134.                     preparedStatement.setString(9, split[8]);//vers
  135.                     preparedStatement.setString(10, split[9]);//rg
  136.                     preparedStatement.setInt(11, Integer.parseInt(split[10]));//rh
  137.                     preparedStatement.setString(12, split[11]);//pg
  138.                     preparedStatement.setInt(13, Integer.parseInt(split[12]));//ph
  139.                     preparedStatement.setString(14, split[13]);//sg
  140.                     preparedStatement.setInt(15, Integer.parseInt(split[14]));//sh
  141.                     String time_stamp = split[15];
  142.                     if ("null".equals(time_stamp) || StringUtils.isBlank(time_stamp)) {
  143.                         time_stamp = "0";
  144.                     }
  145.                     preparedStatement.setLong(16, Long.parseLong(time_stamp));//time_stamp
  146.                     preparedStatement.setString(17, split[16]);//host
  149.                     //rec_feed_ids,txt,vedio,gallery,p_1,p_2,p_3,p_4,p_5,p_6,p_7,p_8,p_9,p_10,p_11,p_12,p_13,p_14,p_15
  150.                     preparedStatement.setString(18, split[17]);//rec_feed_ids
  151.                     preparedStatement.setString(19, split[18]);//txt
  152.                     preparedStatement.setString(20, split[19]);//vedio
  153.                     preparedStatement.setString(21, split[20]);//gallery
  154.                     preparedStatement.setString(22, split[21]);//p_1
  155.                     preparedStatement.setString(23, split[22]);//p_1
  156.                     preparedStatement.setString(24, split[23]);//p_1
  157.                     preparedStatement.setString(25, split[24]);//p_1
  158.                     preparedStatement.setString(26, split[25]);//p_1
  159.                     preparedStatement.setString(27, split[26]);//p_1
  160.                     preparedStatement.setString(28, split[27]);//p_1
  161.                     preparedStatement.setString(29, split[28]);//p_1
  162.                     preparedStatement.setString(30, split[29]);//p_1
  163.                     preparedStatement.setString(31, split[30]);//p_1
  164.                     preparedStatement.setString(32, split[31]);//p_1
  165.                     preparedStatement.setString(33, split[32]);//p_1
  166.                     preparedStatement.setString(34, split[33]);//p_1
  167.                     preparedStatement.setString(35, split[34]);//p_1
  168.                     preparedStatement.setString(36, split[35]);//p_1
  169.                     preparedStatement.setString(37, split[36]);
  170.                     preparedStatement.setString(38, split[37]);
  171.                     preparedStatement.setString(39, split[38]);
  172.                     preparedStatement.setString(40, split[39]);
  173.                     preparedStatement.setString(41, split[40]);
  174.                     preparedStatement.setString(42, split[41]);
  175.                     preparedStatement.setString(43, split[42]);
  176.                     preparedStatement.setString(44, split[43]);
  177.                     preparedStatement.setString(45, split[44]);
  178.                     preparedStatement.setString(46, split[45]);
  179.                     preparedStatement.setString(47, split[46]);
  180.                     preparedStatement.setString(48, split[47]);
  181.                     preparedStatement.setString(49, split[48]);
  182.                     preparedStatement.setString(50, split[49]);
  183.                     preparedStatement.setString(51, split[50]);
  184.                     preparedStatement.setString(52, split[51]);
  185.                     preparedStatement.setString(53, split[52]);
  186.                     preparedStatement.setString(54, split[53]);
  187.                     preparedStatement.setString(55, split[54]);
  188.                     preparedStatement.setString(56, split[55]);
  189.                     preparedStatement.setString(57, split[56]);
  190.                     preparedStatement.setString(58, split[57]);
  191.                     preparedStatement.setString(59, split[58]);
  192.                     preparedStatement.setString(60, split[59]);
  193.                     preparedStatement.setString(61, split[60]);
  194.                     preparedStatement.setString(62, split[61]);
  195.                     preparedStatement.setString(63, split[62]);
  196.                     preparedStatement.setString(64, split[63]);
  197.                     preparedStatement.setString(65, split[64]);
  198.                     preparedStatement.setString(66, split[65]);
  199.                     preparedStatement.setString(67, split[66]);
  200.                     preparedStatement.setString(68, split[67]);
  201.                     preparedStatement.setString(69, split[68]);
  202.                     preparedStatement.setString(70, split[69]);
  203.                     preparedStatement.setString(71, split[70]);
  206.                     preparedStatement.addBatch();
  208.                     logger.info("device:{}",split[0]);
  209.                 }
  211.                 preparedStatement.executeBatch();
  212.                 conn.commit();
  213.             }
  215.             transaction.commit();
  216.         } catch (Throwable e) {
  217.             try {
  218.                 if (transaction != null) {
  219.                     transaction.rollback();
  220.                 }
  221.             } catch (Exception e2) {
  222.                 logger.error("flume transaction rollback error.", e2);
  223.             }
  224.             logger.error("Failed to commit flume transaction," +"Transaction rolled back.", e);
  225.             //Throwables.propagate(e);
  226.         } finally {
  227.             if (transaction != null) {
  228.                 transaction.close();
  229.             }
  231.             if (preparedStatement != null) {
  232.                 try {
  233.                     preparedStatement.close();
  234.                 } catch (SQLException e) {
  235.                     logger.error("statement close error.", e);
  236.                 }
  237.             }
  239.             if (conn != null) {
  240.                 try {
  241.                     conn.close();
  242.                 } catch (SQLException e) {
  243.                     logger.error("connection close error.", e);
  244.                 }
  245.             }
  246.         }
  247.         return result;
  248.     }
  250.     @Override
  251.     public void configure(Context context) {
  252.         String deviceKeys = context.getString("deviceKeys");
  253.         if (StringUtils.isNotBlank(deviceKeys)) {
  254.             String[] split = deviceKeys.split(",");
  255.             for (String deviceKey : split) {
  256.                 deviceKeySet.add(deviceKey);
  257.             }
  258.             logger.info("sink configure deivceKeys:"+deviceKeys);
  259.         } else {
  260.             logger.info("sink configure deivceKeys is empty...");
  261.         }
  262.     }
  263. }

说明：

1）configure方法中，可以从flume的配置文件中读取对应的配置信息。当配置文件修改后，flume框架会自动重新加载，这是就会调用configure方法

2）start和stop方法当flume启动和关闭时进行执行；

3、打包：

使用 maven package 将带把打包成ttbrain-log-flume-MysqlSink-jar-with-dependencies.jar

4、部署：

1）配置flume配置文件：

  1. agent1.sources = ngrinder
  2. agent1.channels = mc2
  3. agent1.sinks = mysql
  6. #source
  7. agent1.sources.ngrinder.type = exec
  8. agent1.sources.ngrinder.command = tail -F /data/logs/ttbrain/ttbrain-recommend-api.log
  9. agent1.sources.ngrinder.channels = mc2
  11. #filter
  12. agent1.sources.ngrinder.interceptors=filt1 filt2 filt3 filt4
  13. agent1.sources.ngrinder.interceptors.filt1.type=regex_filter
  14. agent1.sources.ngrinder.interceptors.filt1.regex=.*recId.*
  15. agent1.sources.ngrinder.interceptors.filt2.type=host
  16. agent1.sources.ngrinder.interceptors.filt2.hostHeader=hostname
  17. agent1.sources.ngrinder.interceptors.filt2.useIP=true
  18. agent1.sources.ngrinder.interceptors.filt3.type=timestamp
  19. agent1.sources.ngrinder.interceptors.filt4.type=com.abc.ttbrain.log.flume.interceptor.MyInterceptor$Builder
  22. #channel2
  23. agent1.channels.mc2.type = file
  24. agent1.channels.mc2.checkpointDir = /data/flume/ckdir/mc2_ck
  25. agent1.channels.mc2.dataDirs = /data/flume/datadir/mc2_data
  27. #sink2
  28. agent1.sinks.mysql.type = com.abc.ttbrain.log.flume.sink.MysqlSink
  29. agent1.sinks.mysql.deviceKeys = 2DCFE0C8-2DD6-4FB7-A2E6-1A210F7C7C07,3F4DA241-B827-4FF8-BB3F-624CFDEDA58D,89C574A2-1E44-468F-9AB4-96737D2FF7F2,f614ca8a42bd121a8bb971d89a078a08267b4df2,A78DB69352E74B744EDD15DE2B91BE40,2C4B1E38-2629-4AF9-BA0E-2DA17C5F3A35,B9D0A82EE8A4BB9B20CEEBB0977A9CFC,3BD40EB0-8171-4481-8B6E-002DB8C6924D,81F37B3C-5E29-40BC-A030-424E137268C2,E59909BC-D62E-4393-9D06-41BF03F81DA9
  30. agent1.sinks.mysql.channel = mc2

说明：

A、agent1.sinks.mysql.type 指定自定义sink类全路径；

B、agent1.sinks.mysql.deviceKeys 是自定义配置信息，可以在自定义sink中的configure方法中获取该信息。

2）将ttbrain-log-flume-MysqlSink-jar-with-dependencies.jar放到flume_home 的lib目录下；

3）启动flume：

  1. nohup flume-ng agent -c /usr/local/apache-flume-1.7.0-bin/conf -f /usr/local/apache-flume-1.7.0-bin/conf/engine-api-log.conf  -n agent1 >/dev/null 2>&1 &

发表评论

表情：
评论列表 （有 0 条评论，683人围观）

还没有评论，来说两句吧...

相关阅读

    相关 Flume-自定义Sink

    自定义Sink Sink不断地轮询Channel中的事件切批量地移除他们，并将这些事件批量写入到存储或索引系统、或被发送到另一个Flume Agent； Sink是完全

    - 日理万妓- 日理万妓/ 0 赞/ 69 阅读