Logstash：导入 RSS feed 数据

Posted 2021-06-28 Elastic 中国社区官方博客

在实际的使用过程中，我们有时希望正对 RSS 的数据来进行导入并进行搜索。在现实的许多微服务中，有很多的数据是根据 RSS feed 的形式来提供的，比如我们常见的点评网站等。那么我们有没有办法把这些数据导入到 Elasticsearch 并进行搜索呢？答案是使用 Logstash 所提供的 RSS input plugin。在今天的文章中，我将使用一个例子来进行展示。

我们首先找到一个 RSS feed 源。在 Elastic 的官方网站中，我们可以看到这样的一个页面 elastic.co/blog。我们打开这个页面：

点击上面的 RSS：

在上面，我们可以看到 RSS 的响应结果。它里面含有 title, description 等信息。这些我们可以使用 RSS input plugin 来进行提取。

 

安装

由于 RSS input plugin 不是 Logstash input plugin 中的一个，我们必须进行手动安装。如果你想知道 Logstash 自带的 input plugin，你可以使用如下的命令：

./bin/logstash-plugin list --group input

上面的命令显示：

logstash-input-azure_event_hubs
logstash-input-beats
└── logstash-input-elastic_agent (alias)
logstash-input-couchdb_changes
logstash-input-elasticsearch
logstash-input-exec
logstash-input-file
logstash-input-ganglia
logstash-input-gelf
logstash-input-generator
logstash-input-graphite
logstash-input-heartbeat
logstash-input-http
logstash-input-http_poller
logstash-input-imap
logstash-input-jms
logstash-input-pipe
logstash-input-redis
logstash-input-rss
logstash-input-s3
logstash-input-snmp
logstash-input-snmptrap
logstash-input-sqs
logstash-input-stdin
logstash-input-syslog
logstash-input-tcp
logstash-input-twitter
logstash-input-udp
logstash-input-unix

RSS input plugin 显然是不在里面。我们可以使用如下的命令来进行安装：

./bin/logstash-plugin install logstash-input-rss

$ ./bin/logstash-plugin install logstash-input-rss
Using JAVA_HOME defined java: /Library/Java/JavaVirtualMachines/jdk-15.0.2.jdk/Contents/Home
WARNING, using JAVA_HOME while Logstash distribution comes with a bundled JDK
Validating logstash-input-rss
Installing logstash-input-rss
Installation successful

一旦安装成功，我们再次使用上面的 list 命令来进行查看，我们会发现 logstash-input-rss 会列入其中。

 

导入 RSS feed 数据

接下来，我们使用上面安装的 RSS input plugin 来导入 https://www.elastic.co/blog/feed 中的数据。我们首先在 Logstash 的安装根目录下创建如下的配置文件：

logstash.conf

input {
  rss {
    url => "https://www.elastic.co/blog/feed"
    interval => 120
    tags => ["rss", "elastic"]
  }
}

filter { 
  mutate { 
    rename => [ "message", "blog_html" ] 
    copy => { "blog_html" => "blog_text" } 
    copy => { "published" => "@timestamp" } 
  } 
  mutate { 
    gsub => [  
      "blog_text", "<.*?>", "", 
      "blog_text", "[\\n\\t]", " " 
    ] 
    remove_field => [ "published", "author" ] 
  }
} 
 
output {
  stdout {
    codec => rubydebug
  }

  elasticsearch { 
    hosts => [ "localhost:9200" ] 
    index => "elastic_blog" 
  }
}

在上面的 input 部分，我们定义了 rss。它的 url 是 https://www.elastic.co/blog/feed。interval 为 120，也就是 120 秒。每隔两分钟运行一次。同时我们添加了一下我们想要的 tags 便于我们的数据搜索。在 filter 部分，我们把 message 重新命名为 blog_html，并同时对字段进行了复制。在 mutate 的部分，我们把所有的 <> 括号里的内容都置换为“”，也就是去掉。同时，也置换 \\n\\t 这样的字符为空 “ ”。最后，我们删除 published 及 author 字段尽管删除 author 字段是不必要的。

我们在 Logstash 的安装目录下运行如下的命令：

./bin/logstash -f logstash.conf

我们在 console 中可以看到：

我们可以看到在 blog_text 中，那些 <> tag 里的内容都被去掉了。

我们可到 Elasticsearch 中查找索引 elastic_blog：

GET elastic_blog/_count

{
  "count" : 20,
  "_shards" : {
    "total" : 1,
    "successful" : 1,
    "skipped" : 0,
    "failed" : 0
  }
}

我们甚至可以对它进行搜索：

 

再过两分钟后，我们再次重新查看被导入的数据：

这次，我们看到 count 变为 40，也就是第一次 20 的一倍。这是由于我们每隔2分钟去重新读取一次数据。如果我们再接着等两分钟，那么数据就会变为60：

好了，在今天的练习中。我们讲述了如何使用 Logstash RSS input plugin 来导入 RSS feed 数据。希望对你的工作有所帮助。