【大数据环境准备】(七)flume 采集

本文涉及的产品
日志服务 SLS,月写入数据量 50GB 1个月
云原生大数据计算服务 MaxCompute,5000CU*H 100GB 3个月
云原生大数据计算服务MaxCompute,500CU*H 100GB 3个月
简介: flume 采集

按照规划,需要采集的用户行为日志文件分布在hadoop102,hadoop103两台日志服务器,故需要在hadoop102,hadoop103两台节点配置日志采集Flume。日志采集Flume需要采集日志文件内容,并对日志格式(JSON)进行校验,然后将校验通过的日志发送到Kafka。
此处可选择TaildirSource和KafkaChannel,并配置日志校验拦截器。
选择TailDirSource和KafkaChannel的原因如下:

1)TailDirSource

TailDirSource相比ExecSource、SpoolingDirectorySource的优势

TailDirSource:断点续传、多目录。Flume1.6以前需要自己自定义Source记录每次读取文件位置,实现断点续传。

ExecSource可以实时搜集数据,但是在Flume不运行或者Shell命令出错的情况下,数据将会丢失。

SpoolingDirectorySource监控目录,支持断点续传。

2)KafkaChannel

采用Kafka Channel,省去了Sink,提高了效率

创建拦截器

(1)创建Maven工程flume-interceptor
(2)创建包:org.apache.flume.interceptor HostInterceptor

(3)在pom.xml文件中添加如下配置

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

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

<build>
    <plugins>
        <plugin>
            <artifactId>maven-compiler-plugin</artifactId>
            <version>2.3.2</version>
            <configuration>
                <source>1.8</source>
                <target>1.8</target>
            </configuration>
        </plugin>
        <plugin>
            <artifactId>maven-assembly-plugin</artifactId>
            <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>

拦截代码

package org.apache.flume.interceptor;

import com.alibaba.fastjson.JSONException;
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.Iterator;
import java.util.List;

public class HostInterceptor implements Interceptor {
   
    /*
     * 通过异常判断是否是json字符串
     * 是:返回true  不是:返回false
     * */
    public  boolean isJSONValidate(String log){
   
        try {
   
            JSONObject.parseObject(log);
            return true;
        }catch (JSONException e){
   
            return false;
        }
    }
    @Override
    public void initialize() {
   

    }

    @Override
    public Event intercept(Event event) {
   

        //1、获取body当中的数据并转成字符串
        byte[] body = event.getBody();
        String log = new String(body, StandardCharsets.UTF_8);
        //2、判断字符串是否是一个合法的json,是:返回当前event;不是:返回null
        if (this.isJSONValidate(log)) {
   
            return event;
        } else {
   
            return null;
        }
    }

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

        Iterator<Event> iterator = list.iterator();

        while (iterator.hasNext()){
   
            Event next = iterator.next();
            if(intercept(next)==null){
   
                iterator.remove();
            }
        }

        return list;
    }

    public static class Builder implements Interceptor.Builder{
   

        @Override
        public Interceptor build() {
   
            return new HostInterceptor();
        }
        @Override
        public void configure(Context context) {
   

        }

    }

    @Override
    public void close() {
   

    }
}

打包

复制文件

[centos@hadoop10 data]$ /data/xsync /data/module/flume/
[centos@hadoop10 data]$ /data/xsync /data/module/flume/lib/flume-interceptor-1.0-SNAPSHOT-jar-with-dependencies.jar
[centos@hadoop10 data]$ vim f1.sh

#!/bin/bash
case $1 in
"start"){
   
        for i in hadoop10 hadoop11 hadoop12
        do
                echo " --------启动 $i 采集flume-------"
                ssh $i "nohup /data/module/flume/bin/flume-ng agent -n a1 -c /data/module/flume/conf/ -f /data/module/flume/job/file_to_kafka.conf >/dev/null 2>&1 &"
        done
};; 
"stop"){
   
        for i in hadoop10 hadoop11 hadoop12
        do
                echo " --------停止 $i 采集flume-------"
                ssh {
   mathJaxContainer[0]}2}' | xargs -n1 kill -9 "
        done
};;
esac

[centos@hadoop10 data]$ chmod 777 fl.sh 
[centos@hadoop10 data]$ ./fl.sh start
相关实践学习
基于MaxCompute的热门话题分析
本实验围绕社交用户发布的文章做了详尽的分析,通过分析能得到用户群体年龄分布,性别分布,地理位置分布,以及热门话题的热度。
SaaS 模式云数据仓库必修课
本课程由阿里云开发者社区和阿里云大数据团队共同出品,是SaaS模式云原生数据仓库领导者MaxCompute核心课程。本课程由阿里云资深产品和技术专家们从概念到方法,从场景到实践,体系化的将阿里巴巴飞天大数据平台10多年的经过验证的方法与实践深入浅出的讲给开发者们。帮助大数据开发者快速了解并掌握SaaS模式的云原生的数据仓库,助力开发者学习了解先进的技术栈,并能在实际业务中敏捷的进行大数据分析,赋能企业业务。 通过本课程可以了解SaaS模式云原生数据仓库领导者MaxCompute核心功能及典型适用场景,可应用MaxCompute实现数仓搭建,快速进行大数据分析。适合大数据工程师、大数据分析师 大量数据需要处理、存储和管理,需要搭建数据仓库?学它! 没有足够人员和经验来运维大数据平台,不想自建IDC买机器,需要免运维的大数据平台?会SQL就等于会大数据?学它! 想知道大数据用得对不对,想用更少的钱得到持续演进的数仓能力?获得极致弹性的计算资源和更好的性能,以及持续保护数据安全的生产环境?学它! 想要获得灵活的分析能力,快速洞察数据规律特征?想要兼得数据湖的灵活性与数据仓库的成长性?学它! 出品人:阿里云大数据产品及研发团队专家 产品 MaxCompute 官网 https://www.aliyun.com/product/odps&nbsp;
相关文章
|
6月前
|
存储 消息中间件 监控
【Flume】Flume在大数据分析领域的应用
【4月更文挑战第4天】【Flume】Flume在大数据分析领域的应用
|
6月前
|
运维 安全 Unix
使用JSch远程部署flume采集点
使用JSch远程部署flume采集点
|
30天前
|
SQL 机器学习/深度学习 分布式计算
大数据-81 Spark 安装配置环境 集群环境配置 超详细 三台云服务器
大数据-81 Spark 安装配置环境 集群环境配置 超详细 三台云服务器
54 1
|
25天前
|
分布式计算 Hadoop 大数据
大数据体系知识学习(一):PySpark和Hadoop环境的搭建与测试
这篇文章是关于大数据体系知识学习的,主要介绍了Apache Spark的基本概念、特点、组件,以及如何安装配置Java、PySpark和Hadoop环境。文章还提供了详细的安装步骤和测试代码,帮助读者搭建和测试大数据环境。
47 1
|
1月前
|
存储 数据采集 分布式计算
Hadoop-17 Flume 介绍与环境配置 实机云服务器测试 分布式日志信息收集 海量数据 实时采集引擎 Source Channel Sink 串行复制负载均衡
Hadoop-17 Flume 介绍与环境配置 实机云服务器测试 分布式日志信息收集 海量数据 实时采集引擎 Source Channel Sink 串行复制负载均衡
41 1
|
3月前
|
存储 数据可视化 数据挖掘
大数据环境下的房地产数据分析与预测研究的设计与实现
本文介绍了一个基于Python大数据环境下的昆明房地产市场分析与预测系统,通过数据采集、清洗、分析、机器学习建模和数据可视化技术,为房地产行业提供决策支持和市场洞察,探讨了模型的可行性、功能需求、数据库设计及实现过程,并展望了未来研究方向。
140 4
大数据环境下的房地产数据分析与预测研究的设计与实现
|
3月前
|
存储 分布式计算 大数据
【Flume的大数据之旅】探索Flume如何成为大数据分析的得力助手,从日志收集到实时处理一网打尽!
【8月更文挑战第24天】Apache Flume是一款高效可靠的数据收集系统,专为Hadoop环境设计。它能在数据产生端与分析/存储端间搭建桥梁,适用于日志收集、数据集成、实时处理及数据备份等多种场景。通过监控不同来源的日志文件并将数据标准化后传输至Hadoop等平台,Flume支持了性能监控、数据分析等多种需求。此外,它还能与Apache Storm或Flink等实时处理框架集成,实现数据的即时分析。下面展示了一个简单的Flume配置示例,说明如何将日志数据导入HDFS进行存储。总之,Flume凭借其灵活性和强大的集成能力,在大数据处理流程中占据了重要地位。
69 3
|
4月前
|
JSON 分布式计算 大数据
MaxCompute操作报错合集之连接环境时,出现报错:TypeError: access_id and secret_access_key,该怎么解决
MaxCompute是阿里云提供的大规模离线数据处理服务,用于大数据分析、挖掘和报表生成等场景。在使用MaxCompute进行数据处理时,可能会遇到各种操作报错。以下是一些常见的MaxCompute操作报错及其可能的原因与解决措施的合集。
|
5月前
|
分布式计算 DataWorks 大数据
MaxCompute产品使用问题之如何同步两个环境的参数
MaxCompute作为一款全面的大数据处理平台,广泛应用于各类大数据分析、数据挖掘、BI及机器学习场景。掌握其核心功能、熟练操作流程、遵循最佳实践,可以帮助用户高效、安全地管理和利用海量数据。以下是一个关于MaxCompute产品使用的合集,涵盖了其核心功能、应用场景、操作流程以及最佳实践等内容。
|
5月前
|
数据采集 存储 分布式计算
Nutch爬虫在大数据采集中的应用案例
Nutch爬虫在大数据采集中的应用案例