Flume1.9.0自定义Sink组件将数据发送至Mysql

news2025/10/31 8:12:04

需求

1、将Flume采集到的日志数据也同步保存到MySQL中一份，但是Flume目前不支持直接向MySQL中写数据，所以需要用到自定义Sink，自定义一个MysqlSink。

2、日志数据默认在Linux本地的/data/log/user.log日志文件中，使用Flume采集到MySQL中到user中。

3、user.log的数据格式如下：

2020-01-01 01:10:23,tom,18,beijing
2020-01-01 01:12:09,jack,20,shanghai
2020-01-01 01:13:17,jessic,15,guangzhou

4、mysql中的user表结构如下：

CREATE TABLE user  (
  id int(11) NOT NULL AUTO_INCREMENT,
  name varchar(255),
  age int(11),
  city varchar(255),
  create_time datetime(0),
  PRIMARY KEY (id)
);

实现 

鉴于此，可以使用 Exec Source + File Channel + Custom Mysql Sink 来实现。官方文档如下：

Exec Source:
https://flume.apache.org/releases/content/1.11.0/FlumeUserGuide.html#exec-source

File Channel:
https://flume.apache.org/releases/content/1.11.0/FlumeUserGuide.html#file-channel

Custom Sink:
https://flume.apache.org/releases/content/1.11.0/FlumeUserGuide.html#custom-sink
https://flume.apache.org/releases/content/1.11.0/FlumeDeveloperGuide.html#sink

创建工程

引入依赖

主要是 flume-ng-core 和 mysql-connector-java 依赖，其他可不引入。

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.example</groupId>
    <artifactId>flume-demo</artifactId>
    <version>1.0-SNAPSHOT</version>

    <properties>
        <maven.compiler.source>8</maven.compiler.source>
        <maven.compiler.target>8</maven.compiler.target>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.apache.flume</groupId>
            <artifactId>flume-ng-core</artifactId>
            <version>1.9.0</version>
        </dependency>
        <dependency>
            <groupId>com.fasterxml.jackson.core</groupId>
            <artifactId>jackson-databind</artifactId>
            <version>2.15.4</version>
        </dependency>
<!--        <dependency>-->
<!--            <groupId>com.alibaba</groupId>-->
<!--            <artifactId>fastjson</artifactId>-->
<!--            <version>2.0.25</version>-->
<!--        </dependency>-->
<!--        <dependency>-->
<!--            <groupId>cn.hutool</groupId>-->
<!--            <artifactId>hutool-core</artifactId>-->
<!--            <version>5.8.27</version>-->
<!--        </dependency>-->
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-api</artifactId>
            <version>1.7.10</version>
        </dependency>
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-log4j12</artifactId>
            <version>1.7.10</version>
        </dependency>
        <dependency>
            <groupId>mysql</groupId>
            <artifactId>mysql-connector-java</artifactId>
            <version>8.0.33</version>
        </dependency>

    </dependencies>

</project>

编写 Custom Sink

package com.example.flumedemo.sink;

import com.google.common.base.Charsets;
import com.google.common.base.Preconditions;
import org.apache.flume.*;
import org.apache.flume.conf.Configurable;
import org.apache.flume.sink.AbstractSink;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.*;
import java.util.ArrayList;
import java.util.List;

/**
 * 自定义Sink，实现将数据写入到mysql。
 * <p>
 * 注意：
 * 1、编写完成打包后，需要把当前jar包和mysql驱动包放到flume下的lib目录下。
 * 2、linux直接连linux上的mysql，最好不要连win上的mysql了，避坑。
 *
 * @author liaorj
 * @date 2024/11/14
 */
public class MySink extends AbstractSink implements Configurable {

    private static final Logger logger = LoggerFactory.getLogger(MySink.class);

    private String mysqlUrl;
    private String username;
    private String password;
    private String tableName;
    //表字段，逗号分割。需要和Event body中的数据对应。
    private String tableFields;


    @Override
    public void configure(Context context) {
        mysqlUrl = context.getString("mysqlUrl");
        Preconditions.checkNotNull(mysqlUrl, "mysqlUrl required");
        username = context.getString("username");
        Preconditions.checkNotNull(username, "username required");
        password = context.getString("password");
        Preconditions.checkNotNull(password, "password required");
        tableName = context.getString("tableName");
        Preconditions.checkNotNull(tableName, "tableName required");
        tableFields = context.getString("tableFields");
        Preconditions.checkNotNull(tableFields, "tableFields required");
    }

    @Override
    public Status process() throws EventDeliveryException {
        Status status = null;

        //开启事务
        Channel ch = getChannel();
        Transaction txn = ch.getTransaction();
        txn.begin();
        Event event = null;
        while (true) {
            event = ch.take();
            if (event != null) {
                break;
            }
        }
        Connection conn = null;
        PreparedStatement stmt = null;
        try {
            //获取body中的数据
            String body = new String(event.getBody(), Charsets.UTF_8);

            //如果这两个数组大小不一样，则抛异常
            String[] bodySplit = body.split(",");
            String[] fieldsSplit = tableFields.split(",");
            if (bodySplit.length != fieldsSplit.length) {
                //字段数对不上
                throw new Exception("the number of tableFields is incorrect");
            }

            //根据字段数生成对应的问号
            List<String> questionMarkList = new ArrayList<>();
            for (int i = 0; i < fieldsSplit.length; i++) {
                questionMarkList.add("?");
            }
            String questionMarks = String.join(",", questionMarkList);

            //生成sql并插入数据
            String formatSql = String.format("insert into %s(%s) values(%s)", tableName, tableFields, questionMarks);
            logger.info("-----formatSql={}", formatSql);
            logger.info("-----mysqlUrl={}, username={}, password={}", mysqlUrl, username, password);
            DriverManager.registerDriver(new com.mysql.cj.jdbc.Driver());
            conn = DriverManager.getConnection(mysqlUrl, username, password);
            stmt = conn.prepareStatement(formatSql);
            for (int i = 0; i < bodySplit.length; i++) {
                stmt.setString(i + 1, bodySplit[i]);
            }
            stmt.executeUpdate();
            txn.commit();
            status = Status.READY;
        } catch (Throwable t) {
            //异常则回滚
            txn.rollback();
            status = Status.BACKOFF;
            if (t instanceof Error) {
                throw (Error) t;
            } else {
                throw new EventDeliveryException(t);
            }
        } finally {
            //关闭事务
            txn.close();
            //关闭PrepareStatement预处理
            if (stmt != null) {
                try {
                    stmt.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }
            //关闭Connection连接
            if (conn != null) {
                try {
                    conn.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }
        }
        return status;
    }
}

打包

mvn clean
mvn package
打包好后，需要把当前jar包和mysql驱动包一起上传到linux上的flume目录下的lib目录中，否则会报错驱动找不到。

配置文件

创建配置文件

然后在flume目录下的conf目录下创建配置文件：file-to-mysql.conf，内容如下，注意mysqlUrl/username/password 要修改成自己的。

# example.conf: A single-node Flume configuration

# Name the components on this agent
a1.sources = r1
a1.sinks = k1
a1.channels = c1

# Describe/configure the source
a1.sources.r1.type = exec
a1.sources.r1.command = tail -F /data/log/user.log

# Describe the sink,custom sink to mysql
a1.sinks.k1.type = com.example.flumedemo.sink.MySink
a1.sinks.k1.mysqlUrl = jdbc:mysql://192.168.163.128:3306/flume_demo?useUnicode=true&characterEncoding=UTF-8&zeroDateTimeBehavior=convertToNull&allowMultiQueries=true
a1.sinks.k1.username = root
a1.sinks.k1.password = toor
a1.sinks.k1.tableName = user
a1.sinks.k1.tableFields = create_time,name,age,city

# Use a channel which buffers events in memory
a1.channels.c1.type = file
a1.channels.c1.checkpointDir = /data/user/checkpointDir
a1.channels.c1.dataDirs = /data/user/dataDirs

# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

启动flume

切换到flume目录，执行：

bin/flume-ng agent --name a1 --conf conf --conf-file conf/file-to-mysql.conf -Dflume.root.logger=INFO,console

测试结果

查看flume控制台日志：

查看mysql user表，已插入数据：

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.coloradmin.cn/o/2241736.html

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈，一经查实，立即删除！

相关文章

Onlyoffice配置一 JWT認證

Onlyoffice配置一 JWT認證

案例 使用官網給c# MVC的例子&#xff0c;主要在版本7.2之後&#xff0c;默認加入JWT認證&#xff0c;docker版本尚且可以在创建的时候使用默认的指令避开&#xff0c;但是在exe版本&#xff0c;即使配置为false&#xff0c;重启之后也会默认开启。 简单说一下如何配置 配置J…
阅读更多...
ZeroSSL HTTPS SSL证书ACMESSL申请3个月证书

ZeroSSL HTTPS SSL证书ACMESSL申请3个月证书

目录 一、引言 二、准备工作 三、申请 SSL 证书 四、证书选型 五、ssl重要性 一、引言 目前免费 Lets Encrypt、ZeroSSL、BuyPass、Google Public CA SSL 证书&#xff0c;一般免费3-6个月。从申请难易程度分析&#xff0c;zerossl申请相对快速和简单&#xff0c;亲测速度非…
阅读更多...
MySql 日期周处理方式

MySql 日期周处理方式

MySql 日期周处理方式 最近在做数仓相关工作&#xff0c;最近遇到 几个问题&#xff0c; 1、计算指定日期是一年中的第几周&#xff0c;周一为周的第一天 2、计算周的开始时间&#xff0c;结束时间 3、计算周对应的年 比如 2023-01-01 WEEKOFYEAR(2023-01-01) 是2022年的52周&…
阅读更多...
STM32 BootLoader 刷新项目 (十) Flash擦除-命令0x56

STM32 BootLoader 刷新项目 (十) Flash擦除-命令0x56

STM32 BootLoader 刷新项目 (十) Flash擦除-命令0x56 1. STM32F407 BootLoader 中的 Flash 擦除功能详解 在嵌入式系统中&#xff0c;BootLoader 的设计是非常关键的部分&#xff0c;它负责引导主程序的启动、升级以及安全管理。而在 STM32F407 等 MCU 上实现 BootLoader&…
阅读更多...
【Homework】【5】Learning resources for DQ Robotics in MATLAB

【Homework】【5】Learning resources for DQ Robotics in MATLAB

Lesson 5 代码-TwoDofPlanarRobot.m 表示一个 2 自由度平面机器人。该类包含构造函数、计算正向运动学模型的函数、计算平移雅可比矩阵的函数&#xff0c;以及在二维空间中绘制机器人的函数。 classdef TwoDofPlanarRobot%TwoDofPlanarRobot - 表示一个 2 自由度平面机器人类…
阅读更多...
Uniapp 引入 Android aar 包 和 Android 离线打包

Uniapp 引入 Android aar 包 和 Android 离线打包

需求&#xff1a; 原生安卓 apk 要求嵌入到 uniapp 中&#xff0c;并通过 uniapp 前端调起 app 的相关组件。 下面手把手教你&#xff0c;从 apk 到 aar&#xff0c;以及打包冲突到如何运行&#xff0c;期间我所遇到的问题都会 一 一 进行说明&#xff0c;相关版本以我文章内为…
阅读更多...
你可以通过以下步骤找到并打开 **Visual Studio 开发者命令提示符**：

你可以通过以下步骤找到并打开 **Visual Studio 开发者命令提示符**：

你可以通过以下步骤找到并打开 Visual Studio 开发者命令提示符&#xff1a; 1. 通过开始菜单查找 打开 开始菜单&#xff08;点击屏幕左下角的 Windows 图标&#xff09;。在搜索框中输入 Developer Command Prompt。你应该看到以下几种选项&#xff08;具体取决于你的 Visu…
阅读更多...
北京大学c++程序设计听课笔记101

北京大学c++程序设计听课笔记101

基本概念 程序运行期间&#xff0c;每个函数都会占用一段连续的内存空间。而函数名就是该函数所占内存区域的起始地址&#xff08;也称“入口地址”&#xff09;。我们可以将函数的入口地址赋给一个指针变量&#xff0c;使该指针变量指向该函数。然后通过指针变量就可以调用这个…
阅读更多...
(时序论文阅读)TimeMixer: Decomposable Multiscale Mixing for Time Series Forecasting

(时序论文阅读)TimeMixer: Decomposable Multiscale Mixing for Time Series Forecasting

来源论文iclr2024 论文地址&#xff1a;https://arxiv.org/abs/2405.14616 源码地址&#xff1a; https://github.com/kwuking/TimeMixer 背景 数据是有连续性&#xff0c;周期性&#xff0c;趋势性的。我们这篇文章主要围绕的是用MLP结构来预测数据的周期性具体为&#xff…
阅读更多...
Springboot 使用EasyExcel导出含图片并设置样式的Excel文件

Springboot 使用EasyExcel导出含图片并设置样式的Excel文件

Springboot 使用EasyExcel导出含图片并设置样式的Excel文件 Excel导出系列目录&#xff1a;★★★★尤其注意&#xff1a;引入依赖创建导出模板类逻辑处理controllerservice 导出效果总结 Excel导出系列目录&#xff1a; 【Springboot 使用EasyExcel导出Excel文件】 【Springb…
阅读更多...
【论文分享】基于街景图像识别和深度学习的针对不同移动能力老年人的街道步行可达性研究——以南京成贤街社区为例

【论文分享】基于街景图像识别和深度学习的针对不同移动能力老年人的街道步行可达性研究——以南京成贤街社区为例

全球老龄化趋势加剧, 许多城市中老年人数量不断增加&#xff0c;而现有街道和社区基础设施往往未能满足其步行安全和便利需求。本次我们给大家带来一篇SCI论文的全文翻译&#xff0c;该论文通过探讨不同步行能力的老年人对城市步行环境的需求&#xff0c;提供了关于如何改善城市…
阅读更多...
LM2 : A Simple Society of Language Models Solves Complex Reasoning

LM2 : A Simple Society of Language Models Solves Complex Reasoning

文章目录 题目摘要简介相关工作方法论实验结果结论局限性 题目 LM2&#xff1a;简单的语言模型社会解决复杂推理问题 论文地址&#xff1a;https://aclanthology.org/2024.emnlp-main.920/ 项目地址&#xff1a; https://github.com/LCS2-IIITD/Language_Model_Multiplex 摘要…
阅读更多...
【因果分析方法】MATLAB计算Liang-Kleeman信息流

【因果分析方法】MATLAB计算Liang-Kleeman信息流

【因果分析方法】MATLAB计算Liang-Kleeman信息流 1 Liang-Kleeman信息流2 MATLAB代码2.1 函数代码2.2 案例参考Liang-Kleeman 信息流（Liang-Kleeman Information Flow）是由 Liang 和 Kleeman 提出的基于信息论的因果分析方法。该方法用于量化变量之间的因果关系，通过计算信息…
阅读更多...
[含文档+PPT+源码等]精品基于springboot实现的原生Andriod手机使用管理软件

[含文档+PPT+源码等]精品基于springboot实现的原生Andriod手机使用管理软件

软件开发环境及开发工具&#xff1a; 数据库管理工具&#xff1a;phpstudy/Navicat或者phpstudy/sqlyog 开发工具&#xff1a;Android Studio 后台管理系统涉及技术&#xff1a; 后台使用框架&#xff1a;Springboot 前端使用技术&#xff1a;Vue,HTML5,CSS3、JavaScript等…
阅读更多...
（三十三）队列（queue）

（三十三）队列（queue）

文章目录 1. 队列&#xff08;queue&#xff09;1.1 定义1.2 函数1.3 习题1.3.1 例题&#xff08;周末舞会&#xff09; 2. 双向队列&#xff08;deque&#xff09;2.1 定义2.2 函数2.3 题目2.3.1 例题&#xff08;打BOSS&#xff09; 1. 队列&#xff08;queue&#xff09; 队…
阅读更多...
常用数据类型

常用数据类型

1.数值类型 分为整型和浮点型 double(3,1)&#xff1b;长度是3&#xff0c;小数点后是1&#xff0c;比如99.5&#xff0c;10.0&#xff0c;20.8 这里的float和double与java中的类似&#xff0c;都是IEEE 754标准的浮点数&#xff0c;精度会丢失&#xff0c;存在一定误差&#…
阅读更多...
Vue3 -- 集成sass【项目集成5】

Vue3 -- 集成sass【项目集成5】

集成sass&#xff1a; 看过博主的 配置styleLint工具应该已经安装过 sass sass-loader 了&#xff0c;所以我们只需要加上我们的 lang"scss"即可。 <style scoped lang"scss"></style>给项目添加全局样式文件&#xff1a; 在src文件夹下创建…
阅读更多...
【云原生系列--Longhorn的部署】

【云原生系列--Longhorn的部署】

Longhorn部署手册 1.部署longhorn longhorn架构图&#xff1a; 1.1部署环境要求 kubernetes版本要大于v1.21 每个节点都必须装open-iscsi &#xff0c;Longhorn依赖于 iscsiadm主机为 Kubernetes 提供持久卷。 apt-get install -y open-iscsiRWX 支持要求每个节点都安装 N…
阅读更多...
Springboot集成ElasticSearch实现minio文件内容全文检索

Springboot集成ElasticSearch实现minio文件内容全文检索

一、docker安装Elasticsearch &#xff08;1&#xff09;springboot和Elasticsearch的版本对应关系如下&#xff0c;请看版本对应&#xff1a; 注意安装对应版本&#xff0c;否则可能会出现一些未知的错误。 &#xff08;2&#xff09;拉取镜像 docker pull elasticsearch:7…
阅读更多...
Diff 算法的误判

Diff 算法的误判

起源&#xff1a; 设想一下&#xff0c;假如你桌面上的文件都没有文件名&#xff0c;取而代之的是&#xff0c;你使用通过文件的位置顺序即index来区分它们———第一个文件&#xff0c;第二个文件&#xff0c;以此类推。也许这种方式可行&#xff0c;可是一旦你删除了其中的一…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023