安装Logstash并导入Movielens测试数据集(基于elasticsearch-8.5.2版本)

news2024/11/25 15:00:09

安装Logstash并导入Movielens测试数据集

0 安装前准备工作

0.1 安装包下载

组件	安装包	下载地址
Logstash	logstash-8.5.2-linux-x86_64.tar.gz	elastic官网：https://www.elastic.co/cn/downloads/past-releases#logstash elastic中文社区：https://elasticsearch.cn/download/
MovieLens 测试数据集	ml-latest-small.zip	https://grouplens.org/datasets/movielens/

0.2 自定义部署标准

自定义部署标准	标准描述
es	elasticsearch不允许使用root账号启动服务，如果你当前账号是root，则需要创建一个专有账户。本文档使用es作为专有用户。
/home/es/software	用于解压安装elasticsearch、java、kibana、logstash的软件安装包和相关配置文件。
/home/es/software/logstash	logstash-8.5.2的软连接
/home/es/software/ml-latest-small	解压后的Movielens测试数据集

0.3 Movielens测试数据集介绍

在这里插入图片描述

/home/es/software/ml-latest-small的目录结构如图上：

links.csv

文件中的字段数据格式为：movieId,imdbId,tmdbId

movieId：每部电影的id

imdbId：电影排行榜

genres：电影类型

movies.csv

文件里包含了一部电影的id和标题，以及该电影的类别，里包含了一些电影来源的链接。

文件中的字段数据格式为：movieId, title, genres

movieId：每部电影的id

title：电影的标题

genres：电影的类别

ratings.csv

文件里面的数据包含了每一个用户对于每一部电影的评分，数据排序的顺序按照userId和movieId排列的。

文件中的字段数据格式为：userId, movieId, rating, timestamp

userId：每个用户的id

movieId：每部电影的id

rating：用户评分，是5星制，按半颗星的规模递增(0.5 stars - 5 stars)

timestamp：时间戳，自1970年1月1日零点后到用户提交评价的时间的秒数。

tags.csv

文件中的字段数据格式为：userId,movieId,tag,timestamp

userId：每个用户的id

movieId：每部电影的id

tag：电影分类

timestamp：时间戳，自1970年1月1日零点后到用户提交评价的时间的秒数。

0.4 安装节点

本次安装复用es集群的中的192.168.168.1节点，所以新建用户、修改环境变量、安装jdk本次都不在意义叙述。如果需使用，请参考《Elasticsearch集群搭建手册及配置详情(基于elasticsearch-8.5.2版本)》文章。

IP	node.name	安装组件
192.168.168.1	node-1	logstash和MovieLens 测试数据集

1 安装 Logstash

1.1 上传Logstash软件包和Movielens数据集

上传logstash软件包和Movielens数据到/home/es/software目录下

scp logstash-8.5.2-linux-x86_64.tar.gz $ip:/home/es/software

scp ml-latest-small.zip $ip:/home/es/software

解压软件包，为logstash创建软连接，并解压Movielens数据集

cd /home/es/software
tar -zxvf logstash-8.5.2-linux-x86_64.tar.gz
rm -f logstash-8.5.2-linux-x86_64.tar.gz
ln -s logstash-8.5.2 logstash

unzip ml-latest-small.zip
chown -R es:es /home/es

1.2 新建movies.csv数据集配置文件

切换到es用户下执行

su - es

cd /home/es/software/logstash/config/

logstash.conf文件重要配置项描述如下：

path：为Movielens数据集movies.csv文件绝对路径

separator：为movies.csv文件中的数据分割符

columns：为movies.csv文件中数据的列名

**hosts：填写es集群任意节点的curl地址。**默认localhost个别情况下可能无法识别，建议修改为es集群任意节点ip或域名。

index：填写Movielens数据集的索引名称。movies索引名称可以随意改动

新建logstash.conf文件，在文件中追加如下配置。path和hosts需要修改为对应的配置项，其他配置项不需要做任何改动。

# 读取数据配置
input {
	file {
		# 导入文件目录地址
		path => "/home/es/software/ml-latest-small/movies.csv"
		start_position => "beginning"
		sincedb_path => "/dev/null"
	}
}
# 过滤 格式化数据配置
filter {
	# 配置读取csv文件
	csv {
		# 设置拆分符为 逗号
		separator => ","
		# 指定csv文件的字段 按顺序匹配
		columns => ["id", "content", "genre"]
	}
	# 对genre字段内容进行处理
	mutate {
		split => {
			"genre" => "|"
		}
		remove_field => ["path", "host", "@timestamp", "message"]
	}
	# 对content字段内容进行处理
	mutate {
		split => ["content", "("]
		# 添加字段 title 
		add_field => {
			"title" => "%{[content][0]}"
		}
		# 添加字段 year
		add_field => {
			"year" => "%{[content][1]}"
		}
	}

	mutate {
		# 对year字段进行类型转换
		convert => {
			"year" => "integer"
		}
		strip => ["title"]
		# 删除字段
		remove_field => ["path", "host", "@timestamp", "message", "content"]
	}

}
# 输出配置
output {
	# es相关配置
	elasticsearch {
		hosts => "http://192.168.168.1:9200"
		index => "movies"
		document_id => "%{id}"
		#user => "elastic"
		#password => "changeme"
	}
	stdout {}
}

1.3 启动 Logstash并导入movies.csv数据

切换到es用户下，启动Logstash并导入movies.csv数据

su - es

cd /home/es/software/logstash/bin/

./logstash -f /home/es/software/logstash/config/logstash.conf

如果数据导入成功，会打印如下格式的内容。反之，则证明数据导入失败，那就需要看具体的报错信息，重新启动Logstash并开始导入数据。

{
        "year" => 2018,
    "@version" => "1",
         "log" => {
        "file" => {
            "path" => "/home/es/software/ml-latest-small/movies.csv"
        }
    },
       "event" => {
        "original" => "193587,Bungo Stray Dogs: Dead Apple (2018),Action|Animation\r"
    },
       "genre" => [
        [0] "Action",
        [1] "Animation"
    ],
          "id" => "193587",
       "title" => "Bungo Stray Dogs: Dead Apple"
}
{
        "year" => 1991,
    "@version" => "1",
         "log" => {
        "file" => {
            "path" => "/home/es/software/ml-latest-small/movies.csv"
        }
    },
       "event" => {
        "original" => "193609,Andrew Dice Clay: Dice Rules (1991),Comedy\r"
    },
       "genre" => [
        [0] "Comedy"
    ],
          "id" => "193609",
       "title" => "Andrew Dice Clay: Dice Rules"
}

查看数据是否导入成功

#kibana页面可以使用如下命令查看集群中的索引
GET /_cat/indices?v

#或者在linux任意节点执行curl命令查看。前提任意节点与es节点网络互通。
curl -s -XGET http://192.168.168.1:9200/_cat/indices?v
health status index  uuid                   pri rep docs.count docs.deleted store.size pri.store.size
green  open   movies Ovg0Oyk5T2OosH0mqUhTSQ   1   1       9743            0      5.2mb          2.6mb

下图可以发现movies索引，证明数据导入成功。

在这里插入图片描述

1.4 如果数据导入失败重启Logstash再次导入

导入失败的原因有很多，可能是logstash.conf文件的path和hosts配置错误，也有可能是文件的权限没有修改为es。这里为大家演示一遍导入第二个索引indextest的方法，以便用于演示重启Logstash再次导入数据的流程。

su - es
#1、将logstash.conf文件中的索引名称 movies 修改为 indextest
$ sed -i 's#"movies"#"indextest"#g' /home/es/software/logstash/config/logstash.conf

#2、验证索引名称是否修改成功
$ grep "index" ../config/logstash.conf
		index => "indextest"

#3、重启Logstash再次导入indextest索引数据
#3.1、jps命令查到Logstash的PID，此时Logstash的PID为21110
$ jps
21110 Logstash
12360 Elasticsearch
23199 Jps
#3.2、关闭Logstash进程。kill -9 PID
$ kill -9 21110
#3.3、将数据导入到indextest索引中
cd /home/es/software/logstash/bin/

./logstash -f /home/es/software/logstash/config/logstash.conf