首页 / 更多教程 / 数据采集的flume架构

数据采集的flume架构

内容导读

互联网集市收集整理的这篇技术教程文章主要介绍了数据采集的flume架构，小编现在分享给大家，供广大互联网技能从业者学习和参考。文章包含7437字，纯文字阅读大概需要11分钟。

内容图文

测试1:

新建一个flume1.conf文件

name

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

source

a1.sources.r1.type = netcat
a1.sources.r1.bind = DAQ102
a1.sources.r1.port = 6666

channel

a1.channels.c1.type = memory
a1.channels.c1.capacity = 10000
a1.channels.c1.transactionCapacity = 1000

sink

a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
a1.sinks.k1.kafka.topic = atguigu
a1.sinks.k1.kafka.bootstrap.servers = DAQ102:9092
a1.sinks.k1.kafka.producer.acks = 1
a1.sinks.k1.useFlumeEventFormat = false

bind

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

测试2:
使用选择器,将不同的数据添加到不同的topic中

拦截器代码:

package com.hybg.daq;

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 MyInterceptor implements Interceptor {

        @Override
public void initialize() {

}

@Override
public Event intercept(Event event) {

    //原理:根据body的数据包含什么内容在header中添加什么内容
    //获取header文件
    Map<String, String> headers = event.getHeaders();
    //获取body文件
    String string = new String(event.getBody(), StandardCharsets.UTF_8);
    //判断是否包含某个文件
    if(string.contains("atguigu")){
        headers.put("topic","atguigu");
    }else if(string.contains("shangguigu")){
        headers.put("topic","shangguigu");
    }else {
        headers.put("topic","other");
    }
    return event;
}

@Override
public List<Event> intercept(List<Event> list) {
    for (Event event : list) {
        intercept(event);
    }

    return list;
}

@Override
public void close() {

}

public static class MyBuilder implements Builder{

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

    @Override
    public void configure(Context context) {

    }
}

}

name

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

source

a1.sources.r1.type = netcat
a1.sources.r1.bind = DAQ102
a1.sources.r1.port = 4444

配置拦截器

a1.sources.r1.interceptors = i1
a1.sources.r1.interceptors.i1.type = com.hybg.daq.MyInterceptor$MyBuilder

channel

a1.channels.c1.type = memory
a1.channels.c1.capacity = 10000
a1.channels.c1.transactionCapacity = 1000

sink

a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
a1.sinks.k1.kafka.topic = other
a1.sinks.k1.kafka.bootstrap.servers = DAQ102:9092,DAQ103:9092,DAQ104:9092
a1.sinks.k1.kafka.producer.acks = 1
a1.sinks.k1.useFlumeEventFormat = false

bind

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

测试三:

name

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

source

a1.sources.source1.type = org.apache.flume.source.kafka.KafkaSource
a1.sources.source1.batchSize = 5000
a1.sources.source1.batchDurationMillis = 2000
a1.sources.source1.kafka.bootstrap.servers = DAQ102:9092,DAQ103:9092,DAQ104:9092
a1.sources.source1.kafka.topics = atguigu,shangguigu,other
a1.sources.source1.kafka.consumer.group.id = customs

channel

a1.channels.c1.type = memory
a1.channels.c1.capacity = 10000
a1.channels.c1.transactionCapacity = 1000

sinks

a1.sinks.k1.type = logger

bind

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

①三个组件都有

name

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

sources

a1.sources.r1.type = netcat
a1.sources.r1.bind = DAQ102
a1.sources.r1.port = 5555

channel

a1.channels.c1.type = org.apache.flume.channel.kafka.KafkaChannel
a1.channels.c1.kafka.bootstrap.servers = DAQ102:9092,DAQ103:9092,DAQ104:9092

a1.channels.c1.kafka.topic = atguigu

a1.channels.c1.parseAsFlumeEvent = false

sink

a1.sinks.k1.type = logger

bind

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

②有source有channel

name

a1.sources = r1
a1.channels = c1

sources

a1.sources.r1.type = netcat
a1.sources.r1.bind = DAQ102
a1.sources.r1.port = 5555

channel

a1.channels.c1.type = org.apache.flume.channel.kafka.KafkaChannel
a1.channels.c1.kafka.bootstrap.servers = DAQ102:9092,DAQ103:9092,DAQ104:9092
a1.channels.c1.kafka.topic = atguigu
a1.channels.c1.parseAsFlumeEvent = false

bind

a1.sources.r1.channels = c1

③有sink有channel

name

a1.channels = c1
a1.sinks=k1

channel

a1.channels.c1.type = org.apache.flume.channel.kafka.KafkaChannel
a1.channels.c1.kafka.bootstrap.servers = DAQ102:9092,DAQ103:9092,DAQ104:9092

a1.channels.c1.kafka.topic = atguigu

a1.channels.c1.parseAsFlumeEvent = false

sink

a1.sinks.k1.type = logger

bind

a1.sinks.r1.channel = c1

数据采集模块:

第一层flume:

name

a1.sources = r1
a1.channels = c1

sources

a1.sources.r1.type = TAILDIR
a1.sources.r1.filegroups = f1

将文件夹中所有的app的文件都进行读取

a1.sources.r1.filegroups.f1 = /opt/module/applog/log/app.*
a1.sources.r1.batchSize = 1000

设置断点续传的文件存储位置

a1.sources.r1.positionFile =/opt/module/flume-1.9.0/position/taildir_position.json

拦截器j

a1.sources.r1.interceptors = i1
a1.sources.r1.interceptors.i1.type = com.hybg.daq.TaildirInterceptor$MyBuilder

channel

a1.channels.c1.type = org.apache.flume.channel.kafka.KafkaChannel
a1.channels.c1.kafka.bootstrap.servers = DAQ102:9092,DAQ103:9092
a1.channels.c1.kafka.topic = topic_log
a1.channels.c1.parseAsFlumeEvent = false

bind

a1.sources.r1.channels = c1

第二层flume:

name

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

source

a2.sources.r1.type = org.apache.flume.source.kafka.KafkaSource
a2.sources.r1.batchSize = 1000
a2.sources.r1.batchDurationMillis = 2000
a2.sources.r1.kafka.bootstrap.servers = DAQ102:9092,DAQ103:9092
a2.sources.r1.kafka.topics = topic_log
a2.sources.r1.useFlumeEventFormat = false

拦截器

设置拦截器后可能会导致错误,出现timestrap不能读取的问题

a2.sources.r1.interceptors = i1
a2.sources.r1.interceptors.i1.type = com.hybg.daq.timeInterceptor$MyBuilder

channel

a2.channels.c1.type = file

文件的

a2.channels.c1.dataDirs = /opt/module/flume-1.9.0/jobs/filechannel
a2.channels.c1.capacity = 1000000
a2.channels.c1.transactionCapacity = 10000
a2.channels.c1.checkpointDir = /opt/module/flume-1.9.0/jobs/checkpointdir
a2.channels.c1.keep-alive = 3

a1.channels.c1.useDualCheckpoints = true

a1.channels.c1. backupCheckpointDir = /otherdiskdir

sinks

a2.sinks.k1.type = hdfs
a2.sinks.k1.hdfs.path = /origin_data/gmall/log/topic_log/%Y-%m-%d
a2.sinks.k1.hdfs.filePrefix = log-
a2.sinks.k1.hdfs.round = false
a2.sinks.k1.hdfs.roundValue = 10
a2.sinks.k1.hdfs.rollSize = 134217728
a2.sinks.k1.hdfs.rollCount = 0

控制输出文件是原生文件。

a2.sinks.k1.hdfs.fileType = CompressedStream
a2.sinks.k1.hdfs.codeC = lzop

bind

a2.sources.r1.channels = c1
a2.sinks.k1.channel = c1

启动脚本2:

!/bin/bash

if [ $# -lt 1 ]
then
echo “<<<<<<<<<<<<<<<<<<<<输入有效参数>>>>>>>>>>>>>>>>>>>>”
echo “{start,stop}”
exit
fi

case $1 in
“start”)
echo “<<<<<<<<<<<<<<<<<<<<向HDFS传递数据>>>>>>>>>>>>>>>>>>>>”
ssh DAQ104 “flume-ng agent -c $FLUME_HOME/conf -f $FLUME_HOME/jobs/flume_2.conf -n a2 -Dflume.root.logger=INFO,console 1>/opt/module/flume-1.9.0 2>&1 &”
;;
“stop”)
echo “<<<<<<<<<<<<<<<<<<<<停止向HDFS传递数据>>>>>>>>>>>>>>>>>>>>”
ssh DAQ104 “ps -ef | grep flume_2.conf | grep -v grep | awk ‘{print $2}’ | xargs kill -9”
;;
*)
echo “<<<<<<<<<<<<<<<<<<<<参数错误>>>>>>>>>>>>>>>>>>>>”
;;
esac
————————————————
版权声明：本文为CSDN博主「海洋饼干1126」的原创文章，遵循CC 4.0 BY-SA版权协议，转载请附上原文出处链接及本声明。
原文链接：https://blog.csdn.net/weixin_44868211/article/details/118104017

原文：https://www.cnblogs.com/HYBG-JXMD/p/14925311.html

内容总结

以上是互联网集市为您收集整理的数据采集的flume架构全部内容，希望文章能够帮你解决数据采集的flume架构所遇到的程序开发问题。如果觉得互联网集市技术教程内容还不错，欢迎将互联网集市网站推荐给程序员好友。

内容备注

版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容，请发送邮件至 gblab@vip.qq.com 举报，一经查实，本站将立刻删除。

内容手机端

扫描二维码推送至手机访问。

本文链接：https://qyyshop.com/info/1328042.html

来源：【匿名】

【上一篇】文件输入输出【下一篇】关于IE的RegExp.exec的问题

更多 ►

【数据采集的flume架构】教程文章相关的互联网学习教程文章

数据采集的flume架构【代码】

测试1:新建一个flume1.conf文件namea1.sources = r1 a1.channels = c1 a1.sinks = k1sourcea1.sources.r1.type = netcat a1.sources.r1.bind = DAQ102 a1.sources.r1.port = 6666channela1.channels.c1.type = memory a1.channels.c1.capacity = 10000 a1.channels.c1.transactionCapacity = 1000sinka1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink a1.sinks.k1.kafka.topic = atguigu a1.sinks.k1.kafka.bootstrap.se...

url采集工具,url采集器【图】

百度url采集工具，批量采集域名。最好用的关键词域名采集工具，是一款批量采集搜索结果中的url域名的工具。神马url采集工具google谷歌url采集工具批量导入关键词采集url，不限制采集层数，可去重复，可按条件过滤无用网址，帮助你快速获取某一类型的大量url网址，尤其适合想短期内从搜索引擎获取高流量的站长。本站自用原创软件，网上绝无下载。软件价格100元，谢绝还价！购买软件请加微信原文：https://www.cnblogs.com/bianke/p/...

第二篇：速卖通产品采集系列之产品采集实战【代码】【图】

上一篇，对速卖通产品采集做了分析，包含要采集产品信息，以及如何采集这些产品信息，这一篇接着来采集实战，相关技术前篇也说过了，不废话直接开项目做。一, 创建解决方案，编写采集代码1. 创建解决方案“CollectorSolution”，在其中新建“Collector” 空 ASP.NET MVC 项目，解决方案结构图如下：2.在“Collector” 项目中，分别新增“CollectingController” 控制器，以及和控制器相关的视图，并将原来默认路由 Home -》 Index ...

脑电采集地电极和参考电极的作用和区别

安置在头皮上的电极为作用电极（active electrode）。记录到的脑电信号即是作用电极与参考电极的差值。　　放置在身体相对零电位点的电极即为参考电极（reference electrode），也称为参考电极或标准电极。　　如果身体上有一个零电位点，那么将参考电极放置于这个点，头皮上其它部位与该点的电极之间的电位差就等于后者的电位变化的绝对值。但这种零电位点理论上指的是机体位于电解质液中时，距离机体无限远的点，而实际上我们能够...

如何批量采集站长素材某个风格页的音效方法【图】

站长素材里提供了网页素材，素材天下，素材库及音效素材等免费的采集，想批量采集最简单的方法不过是用“视频下载高手”来完成，一起来试试吧。进入站长素材，在“风格”中任选一种进入某个风格页，并复制上方的链接打开视频下载高手，选择“音频音效下载” 弹出的页面上选择“ZZ素材音效列表地址”并把链接粘贴进去点“确定”获取到该风格的全部音效在其序号的小方框中勾选想要下载的音效，并单击“立即下载” 状态栏提示...

智能网关--同时采集多台PLC设备的数据到SQL数据库，存入同一个数据表【代码】【图】

项目中有两台S7-1500的PLC，需要将其数据通过一个智能网关采集到MySQL数据库；网关采集多台智能仪表数据到数据库首先用数据库工具navicat连接数据库建立database数据库和devdata数据表，数据表的结构如下图：然后打开智能网关的参数软件，设置网关的网络参数，具体如下图：参数设置完成后下载，重启网关后新的参数生效启用；可以通过工具->查找搜索设备，用网关PING一下PLC和服务器的IP；最后配...

CMDB资产管理采集【代码】

1 import paramiko2 3 private_key = paramiko.RSAKey.from_private_key_file(‘/home/auto/.ssh/id_rsa‘)4 5 # 创建SSH对象6 ssh = paramiko.SSHClient()7 # 允许连接不在know_hosts文件中的主机8 ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())9 # 连接服务器 10 ssh.connect(hostname=‘c1.salt.com‘, port=22, username=‘wupeiqi‘, key=private_key) 11 12 # 执行命令 13 stdin, stdout, stderr = ssh.ex...

CMDB资产采集方式之ssh【图】

CMDB资产采集方式之ssh标签：info cmdb 技术资产采集 ima src png mic img 本文系统来源：https://www.cnblogs.com/jintian/p/11267268.html

读取webbrowser 中的 cookies 到 httpwebrequest，并实现采集实例下载【图】

【实例简介】读取webbrowser 中的cookie，用httpWebRequest模拟抓包【实例截图】文件：590m.com/f/25127180-494083797-31d925（访问密码：551685）以下内容无关： -------------------------------------------分割线--------------------------------------------- 开发者工具的变革从第一台科学计算机出现以来，人机交互的手段就一直在丰富和发展。通过最初的打孔纸卡，计算机可以“读懂”预置的程序逻辑，让计算机进入“可编...

神策数据盛永根：微信生态——全数据采集和打通【图】

本文根据神策数据盛永根《微信生态数字化运营跨端数据采集方案》直播整理而得，本文的主要内容如下：微信生态常见应用介绍微信生态数据采集各应用间的数据打通一、微信生态介绍1. 微信公众平台公众平台主要包含小程序、服务号、订阅号和微信网页。日常运营都是在公众平台进行，比如修改公众号文章，上传小程序等。2. 微信开放平台开放平台通常包含四个方面：移动应用。若 App 要给用户提供微信的分享功能、支付功能等，就需要接入...

Emlog特定主题模板采集插件【图】

介绍：支持采集的模板：FLY主题、lu主题、Meta主题；其他自测！（就是说那个网站用的是这几个主题，就可以采集他网站的文章。）网盘下载地址： https://zijiewangpan.com/vY3DP3xbWta图片：

哈斯 HAAS 数控系统，网口数据采集【图】

1.哈斯HAAS采集，网口版经过项目验证过的，相对于fanuc ,三菱，数据相对较少。2.机床按了【选择性停止】，做到后面换刀时，机器会停止加工，【此时停止采到的状态也是加工】,而真实的状态是待机。需在其它条件进行判断 3.HAAS 取各种时间的坑总的循环时间当前的循环时间转换问题。 4.HAAS采集程序要注意几个地方，不然容易不稳定，程序死掉后，再次链接链不上的情况。网口是项目中验证OK的。串口理论中跟网口发的命令是一样...

个人采集等保测评+护网漏洞利用工具【图】

1.浪潮ClusterEngineV4.0 sysShell文件远程命令执行漏洞批量扫描poc和exp 项目地址：https://github.com/MzzdToT/ClusterEngineV4.0sysShell_rce 2.TSLab-Exploit 项目地址：https://github.com/tangxiaofeng7/TSLab-Exploit 为了挖更多的漏洞，开发了这款工具。 3.NSNGFW网康下一代防火墙RCE 漏洞批量检测&利用&批量getshell 项目地址：https://github.com/S0por/NSNGFW-RCE-POC_EXP 4.银澎云计算好视通视频会议系统存在任意文件...

狂雨小说1.2.2 CMS系统源码带两套优化模板和一套采集规则【图】

介绍：狂雨1.2.2小说CMS系统源码带两套优化模板和一套采集规则，带安装教程网盘下载地址： http://kekewl.org/5hSGXaFgNtk图片：

虾皮shopee ERP系统采集上传刊登货代系统【图】

货代仓储打包系统功能分为四大部分文章出自作者LAU768 第一部分，客户管理给你下边的用户开设客户端账号，绑定客户店铺，同步客户订单，以及客户在线充值打包费物流费，也可设置账号功能的权限！第二部分，充值管理客户充值记录，每一次充值金额大小，累计充值多少，每一笔消费记录，都可以看的一清二楚！第三部分，仓库管理仓库订单，可同步你名下所有ERP用户的订单数据，客户可在客户端ERP选择发货，相应订单填入...

采集 - 相关标签

采集程序

首页 / 更多教程 / 数据采集的flume架构

数据采集的flume架构

内容导读

内容图文

name

source

channel

sink

bind

name

source

配置拦截器

channel

sink

bind

name

source

channel

sinks

bind

name

sources

channel

a1.channels.c1.kafka.topic = atguigu

sink

bind

name

sources

channel

bind

name

channel

a1.channels.c1.kafka.topic = atguigu

sink

bind

name

sources

将文件夹中所有的app的文件都进行读取

设置断点续传的文件存储位置

拦截器j

channel

bind

name

source

拦截器

设置拦截器后可能会导致错误,出现timestrap不能读取的问题

channel

文件的

a1.channels.c1.useDualCheckpoints = true

a1.channels.c1. backupCheckpointDir = /otherdiskdir

sinks

控制输出文件是原生文件。

bind

!/bin/bash

内容总结

内容备注

内容手机端

【数据采集的flume架构】教程文章相关的互联网学习教程文章

采集 - 相关标签

数据 - 相关标签

更多教程 - 最新教程

更多教程 - 最热教程