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

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

0 安装前准备工作

0.1 安装包下载

0.2 自定义部署标准

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版本)》文章。

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
1
2
3

解压软件包，为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
3
4
5
6
7

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

切换到es用户下执行

su - es

cd /home/es/software/logstash/config/
1
2
3

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 => "nul"
  }
}
filter {
  csv {
    separator => ","
    columns => ["id","content","genre"]
  }

  mutate {
    split => { "genre" => "|" }
    remove_field => ["path", "host","@timestamp","message"]
  }

  mutate {

    split => ["content", "("]
    add_field => { "title" => "%{[content][0]}"}
    add_field => { "year" => "%{[content][1]}"}
  }

  mutate {
    convert => {
      "year" => "integer"
    }
    strip => ["title"]
    remove_field => ["path", "host","@timestamp","message","content"]
  }

}
output {
   elasticsearch {
     # 连接elasticserach 并导入数据
     hosts => "http://192.168.168.1:9200"
     index => "movies"
     document_id => "%{id}"
   }
  stdout {}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44

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

1
2
3
4
5
6

如果数据导入成功，会打印如下格式的内容。反之，则证明数据导入失败，那就需要看具体的报错信息，重新启动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"
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

查看数据是否导入成功

#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
1
2
3
4
5
6
7

下图可以发现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
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

此时可以看到indextest，证明重启Logstash再次导入成功。