当前位置: 首页 > article >正文

医疗数仓配置Flume

医疗数仓配置Flume

  • Flume配置概述

Flume配置概述

Flume需要将Kafka中各topic的数据传输到HDFS,因此选用KafkaSource以及HDFSSink。对于安全性要求高的数据(不允许丢失)选用FileChannel,允许部分丢失的数据如日志可以选用MemoryChannel以追求更高的效率。此处采集的是业务数据,不允许丢失,选用FileChannel,生产环境根据实际情况选择合适的组件。

KafkaSource订阅Kafka medical_ods主题的数据,HDFSSink将不同topic的数据写入不同路径,路径中应包含表名及日期,前者用于区分来源于不同业务表的数据,后者按天对数据进行划分。关键配置如下:
在这里插入图片描述
在这里插入图片描述
2)Flume配置实操
(1)创建Flume配置文件
在hadoop104节点的Flume家目录下创建job目录,在job下创建medical_kafka_to_hdfs.conf。

[atguigu@hadoop104 flume]$ mkdir job 
[atguigu@hadoop104 flume]$ cd job/
[atguigu@hadoop104 job]$ vim medical_kafka_to_hdfs.conf 

(2)配置文件内容如下

a1.sources = r1
a1.channels = c1
a1.sinks = k1

a1.sources.r1.type = org.apache.flume.source.kafka.KafkaSource
a1.sources.r1.batchSize = 5000
a1.sources.r1.batchDurationMillis = 2000
a1.sources.r1.kafka.bootstrap.servers = hadoop102:9092,hadoop103:9092,hadoop104:9092
a1.sources.r1.kafka.topics = topic_db
a1.sources.r1.kafka.consumer.group.id = medical-flume
a1.sources.r1.interceptors = i1
a1.sources.r1.interceptors.i1.type = com.atguigu.medical.flume.interceptors.TimestampAndTableNameInterceptor$Builder


a1.channels.c1.type = file
a1.channels.c1.checkpointDir = /opt/module/flume/checkpoint/medical
a1.channels.c1.dataDirs = /opt/module/flume/data/medical
a1.channels.c1.maxFileSize = 2146435071
a1.channels.c1.capacity = 1000000
a1.channels.c1.keep-alive = 6


## sink1
a1.sinks.k1.type = hdfs
a1.sinks.k1.hdfs.path = /origin_data/medical/%{tableName}_inc/%Y-%m-%d
a1.sinks.k1.hdfs.filePrefix = db
a1.sinks.k1.hdfs.round = false


a1.sinks.k1.hdfs.rollInterval = 10
a1.sinks.k1.hdfs.rollSize = 134217728
a1.sinks.k1.hdfs.rollCount = 0


a1.sinks.k1.hdfs.fileType = CompressedStream
a1.sinks.k1.hdfs.codeC = gzip

## 拼装
a1.sources.r1.channels = c1
a1.sinks.k1.channel= c1

(3)编写Flume拦截器
① 创建名为medical-flume-interceptor的项目
在这里插入图片描述
② 在pom文件中添加如下内容。

<dependencies>
    <dependency>
        <groupId>org.apache.flume</groupId>
        <artifactId>flume-ng-core</artifactId>
        <version>1.10.1</version>
        <scope>provided</scope>
    </dependency>

    <dependency>
        <groupId>com.alibaba</groupId>
        <artifactId>fastjson</artifactId>
        <version>1.2.68</version>
    </dependency>
</dependencies>

<build>
    <plugins>
        <plugin>
            <groupId>org.apache.maven.plugins</groupId>
            <artifactId>maven-assembly-plugin</artifactId>
            <version>3.0.0</version>
            <configuration>
                <descriptorRefs>
                    <descriptorRef>jar-with-dependencies</descriptorRef>
                </descriptorRefs>
            </configuration>
            <executions>
                <execution>
                    <id>make-assembly</id>
                    <phase>package</phase>
                    <goals>
                        <goal>single</goal>
                    </goals>
                </execution>
            </executions>
        </plugin>
    </plugins>
</build>

③ 在com.atguigu.medical.flume.interceptors包下创建TimestampAndTableNameInterceptor类

package com.atguigu.medical.flume.interceptors;

import com.alibaba.fastjson.JSONObject;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;

import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.Map;

public class TimestampAndTableNameInterceptor implements Interceptor {
    @Override
    public void initialize() {

    }

    @Override
    public Event intercept(Event event) {

        Map<String, String> headers = event.getHeaders();
        String log = new String(event.getBody(), StandardCharsets.UTF_8);

        JSONObject jsonObject = JSONObject.parseObject(log);

        Long ts = jsonObject.getLong("ts");

        String tableName = jsonObject.getString("table");

        headers.put("timestamp", ts * 1000 + "");
        headers.put("tableName", tableName);
        return event;

    }

    @Override
    public List<Event> intercept(List<Event> events) {

        for (Event event : events) {
            intercept(event);
        }

        return events;
    }

    @Override
    public void close() {

    }

    public static class Builder implements Interceptor.Builder {


        @Override
        public Interceptor build() {
            return new TimestampAndTableNameInterceptor();
        }

        @Override
        public void configure(Context context) {

        }
    }
}

④ 打包
在这里插入图片描述
⑤ 在target目录下查看打好的包
在这里插入图片描述
⑥ 将打好的包放入到hadoop104的/opt/module/flume/lib文件夹下

[atguigu@hadoop104 lib]$ ls | grep medical-flume-interceptor
medical-flume-interceptor-1.0-SNAPSHOT-jar-with-dependencies.jar

3)通道测试
(1)启动Zookeeper、Kafka集群
(2)启动hadoop104的Flume
[atguigu@hadoop104 flume]$ bin/flume-ng agent -n a1 -c conf/ -f job/medical_kafka_to_hdfs.conf -Dflume.root.logger=INFO,console
(3)执行模拟数据jar包,生成数据
[atguigu@hadoop102 bin]$ medical_mock.sh 1
(4)观察HDFS上的目标路径,如下。
在这里插入图片描述
增量表目录已生成。

4)编写Flume启停脚本
为方便使用,此处编写一个Flume的启停脚本
(1)在hadoop102节点的/home/atguigu/bin目录下创建脚本medical-f1.sh
[atguigu@hadoop102 bin]$ vim medical-f1.sh
在脚本中填写如下内容

#!/bin/bash

case $1 in
"start")
        echo " --------启动 hadoop104 业务数据flume-------"
        ssh hadoop104 "nohup /opt/module/flume/bin/flume-ng agent -n a1 -c /opt/module/flume/conf -f /opt/module/flume/job/medical_kafka_to_hdfs.conf > /opt/module/flume/medical-f1.log 2>&1 &"
;;
"stop")

        echo " --------停止 hadoop104 业务数据flume-------"
        ssh hadoop104 "ps -ef | grep medical_kafka_to_hdfs | grep -v grep |awk '{print \$2}' | xargs -n1 kill"
;;
esac

(2)增加脚本执行权限
[atguigu@hadoop102 bin]$ chmod +x medical-f1.sh
(3)medical-f1启动
[atguigu@hadoop102 module]$ medical-f1.sh start
(4)medical-f1停止
[atguigu@hadoop102 module]$ medical-f1.sh stop


http://www.kler.cn/a/460265.html

相关文章:

  • 【AI】最近有款毛茸茸AI生成图片圈粉了,博主也尝试使用风格转换生成可爱的小兔子,一起来探索下是如何实现的
  • HTML 字符编码
  • 游戏引擎学习第69天
  • 【QT】实现RestFul接口
  • el-form+el-date-picker组合使用时候的回显问题
  • Kraft模式安装Kafka(含常规、容器两种安装方式)
  • 使用maven-mvnd替换maven大大提升编译打包速度
  • sublime 文件高亮设置
  • vim编辑器实用设置
  • VirtualBox新版本报错 Invalid installation directory解决方案
  • C#封送类
  • Tesseract-OCR 文字识别
  • 【Spring】Spring DI(依赖注入)详解—自动装配—byType实现原理
  • 智元与汇川加码,机器人如何利好电机市场?
  • Sigrity System SI SerialLink模式进行HDMI2协议仿真分析操作指导-TP1
  • AI安全的挑战:如何让人工智能变得更加可信
  • 【从零开始入门unity游戏开发之——C#篇41】C#迭代器(Iterator)——自定义类实现 foreach 操作
  • 图像处理-Ch7-小波函数
  • 开源大数据平台E-MapReduce
  • 【广州计算机学会、广州互联网协会联合主办 | ACM独立出版 | 高录用】第四届大数据、信息与计算机网络国际学术会议(BDICN 2025)
  • 【电路理论四】正弦电流电路
  • 前端经典面试合集(二)——Vue/React/Node/工程化工具/计算机网络
  • Log4j2的Filters配置详解(ThresholdFilter )
  • ROS自学笔记三十:话题消息输出并转换为Excel形式
  • python钉钉机器人
  • 【探商宝】企业查询多维度解析---创新信息篇