解决方案:离线数仓02——用户行为日志和采集

　　解决方案:离线数仓02——用户行为日志和采集

　　文章目录

　　第三章用户行为日志 3.1 用户行为日志概述

　　用户行为日志的内容主要包括用户的各种行为信息和行为所处的环境信息。采集

这些信息的主要目的是优化产品，为各项分析统计指标提供数据支持。采集

这些信息的手段通常是埋点。

　　目前主流的埋点方式有代码埋点（前端/后端）、可视化埋点、全量埋点等。

　　代码埋点是调用埋点SDK函数，在需要埋点的业务逻辑函数所在位置调用接口，上报埋点数据。比如我们在页面中嵌入了一个按钮，当按钮被点击时，我们可以在按钮对应的OnClick函数中调用SDK提供的数据发送接口发送数据。

　　可视化嵌入只需要研发人员集成采集SDK，不需要编写嵌入代码。业务人员可以接入分析平台的“圈选”功能，对需要抓取用户行为的控件进行“圈选”，并为Events命名。圈选后，这些配置会同步到每个用户终端，采集SDK会根据圈选的配置自动采集并发送用户行为数据。

　　全埋点是在产品中嵌入SDK，前端自动采集页面所有用户行为事件并上报埋点数据，相当于做一个统一的埋点。然后通过界面配置系统中需要分析哪些数据。

　　3.2 用户行为日志内容

　　本项目采集

分析的用户行为信息主要包括页面浏览记录、动作记录、曝光记录、启动记录和错误记录。

　　3.2.1 页面浏览记录

　　页面浏览记录记录了访问者对该页面的浏览行为。该行为的环境信息主要包括用户信息、时间信息、地理位置信息、设备信息、应用信息、渠道信息、页面信息。

　　3.2.2 动作记录

　　动作记录记录了用户的业务操作行为。行为的环境信息主要包括用户信息、时间信息、地理位置信息、设备信息、应用信息、渠道信息、动作目标信息等。

　　3.2.3 暴露记录

　　暴露记录记录暴露行为。该行为的环境信息主要包括用户信息、时间信息、地理位置信息、设备信息、应用信息、渠道信息、曝光对象信息。

　　3.2.4 开始记录

　　启动记录，记录了用户启动应用程序的行为。该行为的环境信息主要包括用户信息、时间信息、地理位置信息、设备信息、应用信息、频道信息、启动类型、开屏广告信息。

　　3.2.5 错误日志

　　启动记录，记录用户在使用应用过程中的报错行为。该行为的环境信息主要包括用户信息、时间信息、地理位置信息、设备信息、应用信息、渠道信息、可能涉及报错的页面信息、动作信息、曝光信息、动作信息。

　　3.3 用户行为日志格式

　　我们的日志结构大致可以分为两类，一类是页面日志，一类是启动日志。

　　3.3.1 页面日志

　　页面日志，以页面浏览为单位，即一条页面浏览记录生成一条页面跟踪日志。一个完整的页面日志包括一个页面浏览记录、该页面的若干用户操作记录、该页面的若干曝光记录和该页面的错误报告。页面日志除了上述行为信息外，还收录

这些行为的各种环境信息，包括用户信息、时间信息、地理位置信息、设备信息、应用信息、渠道信息等。

　　{

　　"common": { – 环境信息

　　"ar": "230000", -- 区号

　　“ba”: “iPhone”, – 手机品牌

　　"ch": "Appstore", – 频道

　　"is_new": "1", -- 是否第一天使用，首次使用当天该字段值为1，24:00后设置为0。

　　“md”: “iPhone 8”, – 手机型号

　　"mid": "YXfhjAYH6As2z9Iq", – 设备id

　　"os": "iOS 13.2.9", – 操作系统

　　"uid": "485", -- 会员id

　　"vc": "v2.1.134" – 应用版本号

　　},

　　“动作”：[{ - 动作（事件）

　　"action_id": "favor_add", -- 动作id

　　"item": "3", – 目标id

　　"item_type": "sku_id", – 目标类型

　　"ts": 05 – 动作时间戳

　　}

　　],

　　“显示”：[{ – 曝光

　　“displayType”: “query”, – 曝光类型

　　"item": "3", – 曝光对象id

　　"item_type": "sku_id", – 曝光对象类型

　　"order": 1, – 出现顺序

　　“pos_id”: 2 – 曝光位置

　　},

　　{

　　"displayType": "促销",

　　"项目": "6",

　　"item_type": "sku_id",

　　“订单”：2，

　　“pos_id”：1

　　},

　　{

　　"displayType": "促销",

　　"item": "9",

　　"item_type": "sku_id",

　　“订单”：3，

　　“pos_id”：3

　　},

　　{

　　"displayType": "推荐",

　　"项目": "6",

　　"item_type": "sku_id",

　　“订单”：4，

　　“pos_id”：2

　　},

　　{

　　“显示类型”：“查询”，

　　"项目": "6",

　　"item_type": "sku_id",

　　“顺序”：5，

　　“pos_id”：1

　　}

　　],

　　"page": { – 页面信息

　　"during_time": 7648, – 以毫秒为单位的持续时间

　　"item": "3", – 目标id

　　"item_type": "sku_id", – 目标类型

　　"last_page_id": "login", -- 最后一页类型

　　"page_id": "good_detail", – 页面ID

　　"sourceType": "promotion" – 来源类型

　　},

　　"err": { -- 错误

　　"error_code": "1234", --错误码

　　"msg": "**************" -- 错误信息

　　},

　　"ts": 23 --跳转到时间戳

　　}

　　3.3.2 启动日志

　　启动日志以启动为单位，一个启动行为生成一个启动日志。完整的启动日志包括启动记录、启动时的报错、启动时的环境信息，包括用户信息、时间信息、地理位置信息、设备信息、应用程序信息、频道信息等。

　　{

　　“常见的”： {

　　“ar”：“370000”，

　　"ba": "荣誉",

　　"ch": "豌豆家",

　　"is_new": "1",

　　"md": "荣耀20s",

　　“中”：“eQF5boERMJFOujcp”，

　　"os": "安卓11.0",

　　“uid”：“76”，

　　“vc”：“v2.1.134”

　　},

　　“开始”： {

　　"entry": "icon", --icon 手机图标通知安装完成后通知安装开始

　　"loading_time": 18803, -- 开始加载时间

　　"open_ad_id": 7, --广告页面ID

　　“open_ad_ms”: 3449, – 广告总播放时长

　　“open_ad_skip_ms”：1989——用户跳过广告的时间点

　　},

　　"err": { -- 错误

　　"error_code": "1234", --错误代码

　　"msg": "************" -- 错误信息

　　},

　　“TS”：00

　　}

　　3.4 服务器和JDK准备 3.4.1 服务器准备

　　参考下面链接分别安装hadoop102、hadoop103、hadoop104三台主机。

　　虚拟机环境搭建

　　3.4.2 SSH无密码登录配置及JDK准备

　　参考下面的链接完成相应的配置

　　SSH无密码登录配置及JDK准备

　　3.4.3 环境变量配置说明

　　Linux环境变量可以配置在多个文件中，如/etc/profile、/etc/profile.d/*.sh、/.bashrc、/.bash_profile等，以上文件的关系和区别说明以下 。

　　bash的运行方式可分为登录shell和非登录shell。

　　比如我们通过终端输入用户名和密码，登录系统后，得到一个登录shell。而当我们执行如下命令ssh hadoop103命令时，在hadoop103中执行的命令是非登录shell。

　　登录shell和非登录shell的区别

　　这两个 shell 的主要区别在于它们在启动时加载不同的配置文件。当登录 shell 启动时，它会加载 /etc/profile、/.bash_profile、/.bashrc。~/.bashrc 在非登录 shell 启动时加载。

　　在加载/.bashrc（实际上是/etc/bashrc加载到/.bashrc）或/etc/profile时，会执行以下代码片段，

　　因此无论是登录shell还是非登录shell，启动时都会加载/etc/profile.d/*.sh中的环境变量。

　　3.5 仿真数据 3.5.1 使用说明

　　1）上传application.yml、gmall2020-mock-log-2021-10-10.jar、path.json、logback.xml到hadoop102的/opt/module/applog目录下

　　(1) 创建applog路径

　　[atguigu@hadoop102 module]$ mkdir /opt/module/applog

　　2）配置文件

　　(2)上传文件到/opt/module/applog目录

　　(1) application.yml文件

　　可以根据需要生成日期对应的用户行为日志。

　　[atguigu@hadoop102 applog]$ vim application.yml

　　修改以下内容

　　# 外部配置打开

logging.config: "./logback.xml"

#业务日期 注意：并不是Linux系统生成日志的日期，而是生成数据中的时间

mock.date: "2020-06-14"

#模拟数据发送模式

#mock.type: "http"

#mock.type: "kafka"

mock.type: "log"

#http模式下，发送的地址

mock.url: "http://hdp1/applog"

#kafka模式下，发送的地址

mock:

kafka-server: "hdp1:9092,hdp2:9092,hdp3:9092"

kafka-topic: "ODS_BASE_LOG"

#启动次数

mock.startup.count: 200

#设备最大值

mock.max.mid: 500000

#会员最大值

mock.max.uid: 100

#商品最大值

mock.max.sku-id: 35

#页面平均访问时间

mock.page.during-time-ms: 20000

#错误概率 百分比

mock.error.rate: 3

#每条日志发送延迟 ms

mock.log.sleep: 10

#商品详情来源 用户查询，商品推广，智能推荐, 促销活动

mock.detail.source-type-rate: "40:25:15:20"

#领取购物券概率

mock.if_get_coupon_rate: 75

#购物券最大id

mock.max.coupon-id: 3

#搜索关键词

mock.search.keyword: "图书,小米,iphone11,电视,口红,ps5,苹果手机,小米盒子"

　　(2)path.json，这个文件用来配置访问路径

　　可根据需求灵活配置用户点击路径。

　　[

{"path":["home","good_list","good_detail","cart","trade","payment"],"rate":20 },

{"path":["home","search","good_list","good_detail","login","good_detail","cart","trade","payment"],"rate":40 },

{"path":["home","mine","orders_unpaid","trade","payment"],"rate":10 },

{"path":["home","mine","orders_unpaid","good_detail","good_spec","comment","trade","payment"],"rate":5 },

{"path":["home","mine","orders_unpaid","good_detail","good_spec","comment","home"],"rate":5 },

{"path":["home","good_detail"],"rate":10 },

{"path":["home" ],"rate":10 }

]

　　（3）logback配置文件

　　日志生成路径可配置，修改内容如下

　　

%msg%n

${LOG_HOME}/app.%d{yyyy-MM-dd}.log

%msg%n

<p>

" />

</p>

　　3）生成日志

　　(1)进入/opt/module/applog路径，执行以下命令

　　[atguigu@hadoop102 applog]$ java -jar gmall2020-mock-log-2021-10-10.jar

　　(2)在/opt/module/applog/log目录下查看生成的日志

　　[atguigu@hadoop102 log]$ ll

　　3.5.2 集群日志生成脚本

　　在hadoop102的/home/atguigu目录下创建一个bin目录，这样脚本就可以在服务器的任意目录下执行。

　　[atguigu@hadoop102 ~]$ echo $PATH

/usr/local/bin:/usr/bin:/usr/local/sbin:/usr/sbin:/home/atguigu/.local/bin:/home/atguigu/bin

　　(1)在/home/atguigu/bin目录下创建脚本lg.sh

　　[atguigu@hadoop102 bin]$ vim lg.sh

　　（2）在脚本中写入如下内容

　　#!/bin/bash

for i in hadoop102 hadoop103; do

echo "========== $i =========="

ssh $i "cd /opt/module/applog/; java -jar gmall2020-mock-log-2021-10-10.jar >/dev/null 2>&1 &"

done

　　笔记：

　　①/opt/module/applog/为jar包和配置文件所在路径

　　②/dev/null代表Linux的空设备文件，写入该文件的内容全部丢失，俗称“黑洞”。

　　stdin 0: 从键盘获取输入 /proc/self/fd/0

　　标准输出1：输出到屏幕（即控制台）/proc/self/fd/1

　　错误输出2：输出到屏幕（即控制台）/proc/self/fd/2

　　(3)修改脚本执行权限

　　[atguigu@hadoop102 bin]$ chmod 777 lg.sh

　　(4)上传jar包和配置文件到hadoop103的/opt/module/applog/路径下

　　(5) 启动脚本

　　[atguigu@hadoop102 module]$ lg.sh

　　(6)分别在hadoop102和hadoop103的/opt/module/applog/log目录下查看生成的数据

　　[atguigu@hadoop102 logs]$ ls

app.2020-06-14.log

[atguigu@hadoop103 logs]$ ls

app.2020-06-14.log

　　第四章用户行为数据采集模块 4.1 数据通道

　　4.2 环境准备 4.2.1 查看集群所有进程的脚本

　　1）在/home/atguigu/bin目录下创建脚本jpsall

　　[atguigu@hadoop102 bin]$ vim jpsall

　　2）在脚本中写入以下内容

　　#!/bin/bash

for host in hadoop102 hadoop103 hadoop104

do

echo =============== $host ===============

ssh $host jps

done

　　保存退出，然后给脚本执行权限

　　[atguigu@hadoop102 bin]$ chmod +x jpsall

　　3）分发/home/atguigu/bin目录，确保自定义脚本在三台机器上都能使用

　　[atguigu@hadoop102 ~]$ xsync /home/atguigu/bin/

　　4.2.2 Hadoop 安装

　　1）安装步骤

　　详见：hadoop安装与搭建

　　2）项目经历

　　(1) 基于项目经验的HDFS存储多目录

　　① 生产环境服务器磁盘状态

　　② 在hdfs-site.xml文件中配置多个目录，注意新挂载磁盘的访问权限。

　　HDFS的DataNode节点保存数据的路径由dfs.datanode.data.dir参数决定，其默认值为file://${hadoop.tmp.dir}/dfs/data。如果服务器有多个磁盘，则修改参数。如果服务器磁盘如上图所示，则该参数修改为如下值。

　　

dfs.datanode.data.dir

file:///dfs/data1,file:///hd2/dfs/data2,file:///hd3/dfs/data3,file:///hd4/dfs/data4

　　注意：每个服务器挂载的磁盘不同，所以每个节点的多目录配置可能不一致。它可以单独配置。

　　(2) 项目经验的集群数据平衡

　　① 节点间数据平衡

　　启用数据平衡命令：

　　start-balancer.sh -threshold 10

　　对于参数10，表示集群中各个节点的磁盘空间利用率相差不超过10%，可以根据实际情况进行调整。

　　停止数据均衡命令：

　　stop-balancer.sh

　　②磁盘之间的数据平衡

　　生成平衡计划（我们只有一个磁盘，不会生成计划）

　　hdfs diskbalancer -plan hadoop103

　　执行平衡的计划

　　hdfs diskbalancer -execute hadoop103.plan.json

　　查看当前平衡任务的执行状态

　　hdfs diskbalancer -query hadoop103

　　取消平衡任务

　　hdfs diskbalancer -cancel hadoop103.plan.json

　　(3) 基于项目经验的Hadoop参数调优

　　HDFS参数调优 hdfs-site.xml

　　*敏*感*词*客户端请求的 Namenode RPC 服务器线程数。如果 dfs.namenode.servicerpc-address 没有配置，那么 Namenode RPC 服务器线程会*敏*感*词*来自所有节点的请求。

　　NameNode有一个工作线程池来处理不同DataNode的并发心跳和客户端的并发元数据操作。

　　对于大型集群或拥有大量客户端的集群，通常需要增加参数 dfs.namenode.handler.count 的默认值 10。

　　

dfs.namenode.handler.count

10

　　dfs.namenode.handler.count=

　　例如当簇大小为8时，这个参数设置为41。这个值可以通过简单的python代码计算得到，代码如下。

　　[atguigu@hadoop102 ~]$ python

Python 2.7.5 (default, Apr 11 2018, 07:36:10)

[GCC 4.8.5 20150623 (Red Hat 4.8.5-28)] on linux2

Type "help", "copyright", "credits" or "license" for more information.

　　>>> import math

>>> print int(20*math.log(8))

41

>>> quit()

　　YARN 参数调优 yarn-site.xml

　　场景描述：共7台机器，每天上亿条数据，数据源-&gt;Flume-&gt;Kafka-&gt;HDFS-&gt;Hive

　　面临问题：HiveSQL主要用于数据统计，不存在数据倾斜，小文件已合并，启用JVM复用，不阻塞IO，内存使用率低于50%。但是还是跑的很慢，当数据量达到峰值的时候，整个集群就会宕机。基于这种情况，有什么优化方案吗？

　　解决方案：

　　内存利用率不足。这一般是Yarn的两个配置，单个任务可以申请的最大内存大小，以及单个Hadoop节点的可用内存大小导致的。调整这两个参数可以提高系统内存的利用率。

　　(a) yarn.nodemanager.resource.memory-mb

　　表示 YARN 可以在此节点上使用的物理内存总量。默认值为 8192 (MB)。注意，如果你的节点内存资源不足8GB，需要减小这个值，YARN不会智能检测节点总量的物理内存。

　　(b) yarn.scheduler.maximum-allocation-mb

　　单个任务可申请的最大物理内存量，默认为8192（MB）。

　　4.2.3 动物园管理员安装

　　1）安装步骤

　　详见：zookeeper安装步骤

　　4.2.4 Kafka安装

　　1）安装步骤

　　详见：Kafka安装部署

　　4.2.5 Flume安装

　　根据采集通道规划，需要在hadoop102、hadoop103、hadoop104三个节点分别部署一个Flume。可以参考以下步骤先在hadoop102上安装，然后再分发。

　　1）安装步骤

　　详见：Flume安装部署

　　2）分发水槽

　　[atguigu@hadoop102 ~]$ xsync /opt/module/flume/

　　3）项目经历

　　(1)堆内存调整

　　Flume堆内存通常设置为4G或更高，配置如下：

　　修改/opt/module/flume/conf/flume-env.sh文件，配置如下参数（虚拟机环境暂未配置）

　　export JAVA_OPTS="-Xms4096m -Xmx4096m -Dcom.sun.management.jmxremote"

　　笔记：

　　-Xms 表示JVM Heap（堆内存）的最小大小，初始分配；

　　-Xmx表示JVM Heap（堆内存）的最大允许大小，按需分配。

　　4.3 日志采集Flume 4.3.1 日志采集Flume配置概述

　　根据规划，需要采集的用户行为日志文件分布在hadoop102和hadoop103两台日志服务器上，因此需要在hadoop102和hadoop103两个节点上配置日志采集Flume。日志采集

Flume需要采集

日志文件的内容，校验日志格式（JSON），然后将校验过的日志发送给Kafka。

　　这里可以选择TaildirSource和KafkaChannel，配置日志校验*敏*感*词*。

　　选择TailDirSource和KafkaChannel的原因如下：

　　1) 尾目录源

　　TailDirSource 相对于 ExecSource 和 SpoolingDirectorySource 的优势

　　TailDirSource：断点续传，多目录。在Flume 1.6之前，需要自定义Source来记录每次读取文件的位置，从而实现断点续传。

　　ExecSource可以实时采集数据，但是如果Flume没有运行或者Shell命令失败，数据就会丢失。

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

　　2) 卡夫卡通道

　　使用Kafka Channel可以节省Sink，提高效率。

　　日志采集Flume关键配置如下：

　　4.3.2 日志采集Flume配置实践

　　1）创建Flume配置文件

　　在hadoop102节点上Flume的job目录下创建file_to_kafka.conf

　　[atguigu@hadoop104 flume]$ mkdir job

[atguigu@hadoop104 flume]$ vim job/file_to_kafka.conf

　　2）配置文件内容如下

　　#定义组件

a1.sources = r1

a1.channels = c1

#配置source

a1.sources.r1.type = TAILDIR

a1.sources.r1.filegroups = f1

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

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

a1.sources.r1.interceptors = i1

a1.sources.r1.interceptors.i1.type = com.atguigu.gmall.flume.interceptor.ETLInterceptor$Builder

#配置channel

a1.channels.c1.type = org.apache.flume.channel.kafka.KafkaChannel

a1.channels.c1.kafka.bootstrap.servers = hadoop102:9092,hadoop103:9092

a1.channels.c1.kafka.topic = topic_log

a1.channels.c1.parseAsFlumeEvent = false

#组装

a1.sources.r1.channels = c1

　　3）写一个Flume*敏*感*词*

　　（1）创建Maven项目flume-interceptor

　　(2) 创建包：com.atguigu.gmall.flume.interceptor

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

　　

org.apache.flume

flume-ng-core

1.9.0

<p>

" />

provided

com.alibaba

fastjson

1.2.62

maven-compiler-plugin

2.3.2

1.8

1.8

maven-assembly-plugin

jar-with-dependencies

make-assembly

package

single

</p>

　　（4）在com.atguigu.gmall.flume.utils包下创建JSONUtil类

　　package com.atguigu.gmall.flume.utils;

import com.alibaba.fastjson.JSONObject;

import com.alibaba.fastjson.JSONException;

public class JSONUtil {

/*

* 通过异常判断是否是json字符串

* 是：返回true 不是：返回false

* */

public static boolean isJSONValidate(String log){

try {

JSONObject.parseObject(log);

return true;

}catch (JSONException e){

return false;

}

}

}

　　(5)在com.atguigu.gmall.flume.interceptor包下创建一个ETLInterceptor类

　　package com.atguigu.gmall.flume.interceptor;

import com.atguigu.gmall.flume.utils.JSONUtil;

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

@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 (JSONUtil.isJSONValidate(log)) {

return event;

} else {

return null;

}

}

@Override

public List intercept(List list) {

Iterator 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 ETLInterceptor();

}

@Override

public void configure(Context context) {

}

}

@Override

public void close() {

}

}

　　(6) 包装

　　(7)需要将打包好的包放入hadoop102的/opt/module/flume/lib文件夹下。

　　4.3.3 Flume测试日志采集

　　1）启动Zookeeper和Kafka集群

　　2）启动hadoop102的日志采集

Flume

　　[atguigu@hadoop102 flume]$ bin/flume-ng agent -n a1 -c conf/ -f job/file_to_kafka.conf -Dflume.root.logger=info,console

　　3）启动一个Kafka Console-Consumer

　　[atguigu@hadoop102 kafka]$ bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic topic_log

　　4）生成仿真数据

　　[atguigu@hadoop102 ~]$ lg.sh

　　5）观察Kafka消费者是否可以消费数据

　　4.3.4 Flume日志采集启停脚本

　　1）分发日志采集

Flume配置文件和*敏*感*词*

　　如果以上测试通过，则需要将hadoop102节点的Flume配置文件和*敏*感*词*jar包拷贝一份到另一台日志服务器。

　　[atguigu@hadoop102 flume]$ scp -r job hadoop103:/opt/module/flume/

[atguigu@hadoop102 flume]$ scp lib/flume-interceptor-1.0-SNAPSHOT-jar-with-dependencies.jar hadoop103:/opt/module/flume/lib/

　　2）为了方便，这里有一个启动和停止日志采集

Flume进程的脚本

　　在hadoop102节点的/home/atguigu/bin目录下创建脚本f1.sh

　　[atguigu@hadoop102 bin]$ vim f1.sh

　　在脚本中填写以下内容

　　#!/bin/bash

case $1 in

"start"){

for i in hadoop102 hadoop103

do

echo " --------启动 $i 采集flume-------"

ssh $i "nohup /opt/module/flume/bin/flume-ng agent -n a1 -c /opt/module/flume/conf/ -f /opt/module/flume/job/file_to_kafka.conf >/dev/null 2>&1 &"

done

};;

"stop"){

for i in hadoop102 hadoop103

do

echo " --------停止 $i 采集flume-------"

ssh $i "ps -ef | grep file_to_kafka | grep -v grep |awk &#39;{print \$2}&#39; | xargs -n1 kill -9 "

done

};;

esac

　　3）增加脚本执行权限

　　[atguigu@hadoop102 bin]$ chmod 777 f1.sh

　　4）f1开始

　　[atguigu@hadoop102 module]$ f1.sh start

　　5）f1停止

　　[atguigu@hadoop102 module]$ f1.sh stop

　　启动和停止 lume 进程的脚本

　　在hadoop102节点的/home/atguigu/bin目录下创建脚本f1.sh

　　[atguigu@hadoop102 bin]$ vim f1.sh

　　在脚本中填写以下内容

　　#!/bin/bash

case $1 in

"start"){

for i in hadoop102 hadoop103

do

echo " --------启动 $i 采集flume-------"

ssh $i "nohup /opt/module/flume/bin/flume-ng agent -n a1 -c /opt/module/flume/conf/ -f /opt/module/flume/job/file_to_kafka.conf >/dev/null 2>&1 &"

done

};;

"stop"){

for i in hadoop102 hadoop103

do

echo " --------停止 $i 采集flume-------"

ssh $i "ps -ef | grep file_to_kafka | grep -v grep |awk &#39;{print \$2}&#39; | xargs -n1 kill -9 "

done

};;

esac

　　3）增加脚本执行权限

　　[atguigu@hadoop102 bin]$ chmod 777 f1.sh

　　4）f1开始

　　[atguigu@hadoop102 module]$ f1.sh start

　　5）f1停止

　　[atguigu@hadoop102 module]$ f1.sh stop

　　解决方法:优采云

采集器采集文章重复解决办法

　　优采云

采集

者采集

重复文章的解决方法

　　2022年3月25日教程大全优采云

,优采云

collector

　　这两天，我们重新启动了一个垃圾站的采集

工作。发现一些问题，采集结果总是重复两次。采集

器默认设置中的“排除重复设置”不起作用。

　　

" />

　　经过研究，发现可以在内容采集规则中剔除重复，达到效果。

　　建议使用标题来排除重复项。

　　实际效果如下

　　

" />

　　自动删除重复的内容。数据库中的视图没有重复项。

　　柴郡猫发表