2. 程式人生 > 其它 >資料採集的flume架構

資料採集的flume架構

阿新 • • 發佈：2021-06-24

測試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

相關推薦

資料採集的flume架構

測試1: 新建一個flume1.conf檔案 name a1.sources = r1 a1.channels = c1 a1.sinks = k1 source a1.sources.r1.type = netcat

網站流量日誌分析（資料 採集之 Flume 採集）

目錄網站流量日誌分析的意義如何進行網站分析流量分析內容導航分析轉化分析（漏斗模型分析）網站流量日誌分析的資料處理流程資料採集資料預處理資料入庫資料分析資料視覺化埋點資料採集資料採集方式及其優缺點埋點js

大資料實戰（十五）：電商數倉（八）之使用者行為資料採集（八）元件安裝（四）採集日誌Flume

0 簡介 Flume 採集 1日誌採集Flume安裝 叢集規劃： 伺服器hadoop102 伺服器hadoop103 伺服器hadoop104

大資料實戰（十七）：電商數倉（十）之使用者行為資料採集（十）元件安裝（六）消費Kafka資料Flume

0 簡介 叢集規劃 伺服器hadoop102 伺服器hadoop103 伺服器hadoop104 Flume（消費Kafka） Flume

Flume+Kafka+Spark Streaming實現大資料實時流式資料採集

大資料實時流式資料處理是大資料應用中最為常見的場景，與我們的生活也息息相關，以手機流量實時統計來說，它總是能夠實時的統計出使用者的使用的流量，在第一時間通知使用者流量的使用情況，並且最為人性化的為使用

Linux下資料採集工具Flume的安裝

技術標籤：大資料 Flume安裝需要依賴於jdk，首先使用java -version來檢查本機是否將jdk安裝到位

大資料應用-Flume+HBase+Kafka整合資料採集/儲存/分發完整流程測試03.

技術標籤：大資料專欄 系列文章目錄 Flume+HBase+Kafka整合資料採集/儲存/分發完整流程測試

藍橋樓賽第24期-大資料-使用Flume採集資料 題解

技術標籤：hadoopflume題解hadoopflumehdfs 挑戰介紹 在 Hadoop 處理資料之前，首先需要採集資料並且上傳到叢集中。本次挑戰需要你使用 Flume 上傳資料，來監測指定目錄中檔案的變化，一旦該目錄有新檔案時，就會

實時流計算---資料採集工具Flume

Flume最早是Cloudera提供的一個高可用的，高可靠的，分散式的海量日誌採集、聚合和傳輸的系統。

米爾MYC-JX8MMA7核心板丨基於ARM+FPGA異核架構的工業資料與資料採集應用板卡

隨著通訊與網路技術、網際網路的發展，工業管理資料化、網路化、智慧化已成大勢所趨，利用工業物聯網完成工業控制是智慧工廠中必不可少的一部分。傳統的控制與資料採集系統，主機一旦需要同時與多個數據採集裝置保持

美團DB資料同步到資料倉庫的架構與實踐

背景 在資料倉庫建模中，未經任何加工處理的原始業務層資料，我們稱之為ODS(Operational Data Store)資料。在網際網路企業中，常見的ODS資料有業務日誌資料（Log）和業務DB資料（DB）兩類。對於業務DB資料來說，從My

Python實現實時資料採集新型冠狀病毒資料例項

Python實時資料採集-新型冠狀病毒 原始碼 來源：https://github.com/Programming-With-Love/2019-nCoV

.NET Core使用AngleSharp網路資料採集

環境： vs2019 .net core 3.1 angleSharp winform 安裝：angleSharp 有這麼一本Python的書: <<Python 網路資料採集>>

c#學習心得（5）資料採集上位機開發

寫在前面： 紀念一下2020/7/21第一個使用C#開發的上位機專案整體框架成型，具體是採集感測器的力，並計算，實時顯示力曲線，並將資料儲存，

第十講：企業級監控資料採集指令碼開發實踐

　　第十講內容 　　• pushgateway 的介紹 　　• pushgateway 的安裝和運⾏和配置

九齒耙(Ninerake)資料採集大資料深度學習智慧分析Python爬蟲軟體的正則表示式規則簡介

正則表示式易於使用，功能強大，可用於複雜的搜尋和替換以及基於模板的文字檢查。這對於輸入形式的使用者輸入驗證特別有用-驗證電子郵件地址等。您還可以從網頁或文件中提取電話號碼，郵政編碼等，在日誌檔案中搜索複

電商數倉(1)【資料採集】

目錄一.日誌資料採集第一層flume：flume採集資料到kafka1.攔截器（過濾非Json格式資料）2.flume配置檔案3. 啟動flume採集資料第二層flume：從kafka消費資料，寫入到hdfs指定目錄1.攔截器（根據消費資料，設定event h

大資料實戰（十）：電商數倉（三）之使用者行為資料採集（三）資料生成模組（一）

1 埋點資料基本格式 l公共欄位：基本所有安卓手機都包含的欄位 l業務欄位：埋點上報的欄位，有具體的業務型別

大資料實戰（十二）：電商數倉（五）之使用者行為資料採集（五）元件安裝（一）之hadoop安裝

1）叢集規劃： 伺服器hadoop102 伺服器hadoop103 伺服器hadoop104 HDFS NameNode DataNode DataNode

大資料實戰（十三）：電商數倉（六）之使用者行為資料採集（六）元件安裝（二）Zookeeper安裝

1 安裝ZK 叢集規劃 伺服器hadoop102 伺服器hadoop103 伺服器hadoop104 Zookeeper Zookeeper Zookeeper