使用Logstash(流式数据传输)

MaxCompute支持将开源Logstash收集的日志数据写入MaxCompute。您可以通过Logstash的输出插件logstash-output-maxcompute,将Logstash收集的日志数据使用MaxCompute流式数据通道(Streaming Tunnel)功能上传到MaxCompute。

前提条件

在执行操作前请确认您已完成如下操作:

背景信息

Logstash是一个开源的服务器端数据处理管道,可以同时从多个数据源获取数据,并对数据进行转换,然后将转换后的数据发送到用户的目标“存储端”。

您需要通过Logstash的logstash-output-maxcompute插件,将Logstash收集的日志数据使用MaxCompute流式数据通道(Streaming Tunnel)功能上传到MaxCompute。

logstash-output-maxcompute插件基于Logstash v7.8.0版本开发,可以作为输出端口。该插件的特点如下:

  • 使用流式数据通道,避免通过批量数据通道导入产生的并发和小文件问题。

  • 支持动态分区,可以根据Logstash解析的日志字段产生分区字段,能够自动创建不存在的分区。

logstash-output-maxcompute插件应用于如下场景:

  • 需要收集的应用的日志格式在Logstash上有输入插件支持或易于解析,例如NGINX日志。

  • 希望根据日志内容自动创建并导入对应分区。

logstash-output-maxcompute插件支持的数据类型为:STRING、BIGINT、DOUBLE、DATETIME和BOOLEAN。

说明
  • 日志中DATETIME类型的字段的格式将自动使用ruby Time.parse函数推断。

  • 如果日志BOOLEAN字段满足.to_string().lowercase() == "true",则结果为True。其他任何值为False。

本文将以收集NGINX日志为例,介绍如何配置和使用插件。

步骤一:下载并安装插件

您可以下载已安装logstash-output-maxcompute插件的Logstash实例Logstash实例下载链接,跳过安装步骤执行下一步。如果需要自行安装,请按照如下步骤操作:

  1. 下载logstash-output-maxcompute插件(logstash-output-maxcompute插件下载链接)并放置在Logstash的根目录%logstash%下。

  2. 修改Logstash根目录%logstash%下的Gemfile文件,将source "https://rubygems.org"替换为source 'https://gems.ruby-china.com'

  3. 以Windows系统为例,在系统的命令行窗口,切换至Logstash的根目录%logstash%下,执行如下命令安装logstash-output-maxcompute插件。

    bin\logstash-plugin install logstash-output-maxcompute-1.1.0.gem

    当返回Installation successful提示信息时,表示插件安装成功。安装成功

  4. 可选:运行如下命令验证安装结果。

    bin\logstash-plugin list maxcompute
    说明

    Linux系统需要执行命令bin/logstash-plugin list maxcompute

    如果安装成功,会返回logstash-output-maxcompute信息。如果安装失败,解决方案请参见RubyGems验证结果

步骤二:创建目标表

通过MaxCompute客户端或其他可以运行MaxCompute SQL的工具执行如下命令,在目标MaxCompute项目中创建目标表,例如logstash_test_groknginx。后续会将日志信息以日期为分区导入此表中。

create table logstash_test_groknginx(
 clientip string,
 remote_user string,
 time datetime,
 verb string,
 uri string,
 version string,
 response string,
 body_bytes bigint,
 referrer string,
 agent string
) partitioned by (pt string);

步骤三:编写Logstash Pipeline配置文件

在Logstash的根目录%logstash%下创建配置文件pipeline.conf,并输入如下内容:

input { stdin {} }

filter {
        grok {
                match => {
                        "message" => "%{IP:clientip} - (%{USER:remote_user}|-) \[%{HTTPDATE:httptimestamp}\] \"%{WORD:verb} %{NOTSPACE:request} HTTP/%{NUMBER:httpversion}\" %{NUMBER:response} %{NUMBER:body_bytes} %{QS:referrer} %{QS:agent}"
                }
        }
        date {
                match => [ "httptimestamp" , "dd/MMM/yyyy:HH:mm:ss Z" ]
                target => "timestamp"
        }
}

output {
       maxctunnel {
                aliyun_access_id => "<your_accesskey_id>"
                aliyun_access_key => "<your_accesskey_secret>"
                aliyun_mc_endpoint => "<your_project_endpoint>"
                project => "<your_project_name>"
                table => "<table_name>"
                partition => "pt=$<timestamp.strftime('%F')>"
                value_fields => ["clientip", "remote_user", "timestamp", "verb", "request", "httpversion", "response", "bytes", "referrer", "agent"]
        }
}

参数

说明

your_accesskey_id

可以访问目标MaxCompute项目的AccessKey ID。

your_accesskey_secret

AccessKey ID对应的AccessKey Secret。

your_project_endpoint

目标MaxCompute项目所在区域的Endpoint信息。更多Endpoint信息,请参见Endpoint

your_project_name

目标MaxCompute项目的名称。

table_name

目标表的名称,即步骤二中创建的表。

partition

配置插件如何根据日志字段生成对应的分区信息。如果目标表有多个分区,需要指定到最后一级。配置格式如下:

  • 如果某个分区的值为常量,格式为{分区列名}={常量值}

  • 如果某个分区的值为解析后的日志中一个字段的值,格式为{分区列名}=$<{日志字段名}>

  • 如果某个分区的值为解析后的日志中一个日期时间字段的值,并且需要进行重新格式化,格式为{分区列名}=$<{日志字段名}.strftime('{时间格式}')>。其中:{时间格式}是重新格式化的格式化字符串

    在本示例中,将格式化到仅保留日期(%F)。如果要按照日期date作为第一级分区,小时hour作为第二级分区,配置格式为"date=$<timestamp.strftime('%F')>,hour=$<timestamp.strftime('%H')>"

  • 多级分区之间用英文逗号(,)连接,分区指定的顺序和建表时的顺序必须一致。

partition_time_format

可选。指定当一个字符串型的日期时间字段被分区信息引用时,该字段的源格式字符串

在本例中,时间字段timestamp已经被date插件转换为时间类型,因此无需指定。

即使未使用date过滤插件进行转换,亦未指定此配置项的值,在大多数情况下插件仍然可以自动识别内容为日期时间的字符串,并自动完成需要的转换。即只在少数自动识别失败的情况下需要手动指定此项的值。

如果不使用date过滤插件,而是手动进行转换,则需要配置如下信息:

  • 手动指定partition_time_formatpartition_time_format => "%d/%b/%Y:%H:%M:%S %z"

  • 将分区中引用的字段改为日志中的字符串字段:partition => "pt=$<httptimestamp.strftime('%F')>"

value_fields

指定目标表中的每个字段对应的日志字段,指定顺序与表中字段的顺序一致。

目标表字段的顺序为clientip string、remote_user string、time datetime、verb string、uri string、version string、response string、body_bytes bigint、referrer string、agent string,依次对应"clientip"、"remote_user"、"timestamp"、"verb"、"request"、"httpversion"、"response"、"bytes"、"referrer"、"agent"

aliyun_mc_tunnel_endpoint

可选。您可以通过此配置项强制指定Tunnel Endpoint,覆盖自动路由机制。

retry_time

失败重试次数。当写入MaxCompute失败时,尝试重新写入的次数。默认值为3。

retry_interval

失败重试间隔。在两次尝试之间最少间隔的时间,单位为秒。默认值为1。

batch_size

一次最多处理的日志条数。默认值为100。

batch_timeout

写入MaxCompute的超时时间,单位为秒。默认值为5。

说明

在本配置文件中,指定的日志输入为标准输入(input { stdin {} })。在实际应用场景中,您可以使用Logstash File输入插件从本地硬盘中自动读取NGINX日志。更多信息,请参见Logstash文档

步骤四:运行和测试

  1. 以Windows系统为例,在系统的命令行窗口,切换至Logstash的根目录%logstash%下,执行如下命令启动Logstash。

    bin\logstash -f pipeline.conf

    返回Successfully started Logstash API endpoint信息时,Logstash启动完毕。启动完毕

  2. 在系统的命令行窗口,粘贴如下日志样例,并按下键盘上的Enter键。

    1.1.1.1 - - [09/Jul/2020:01:02:03 +0800] "GET /masked/request/uri/1 HTTP/1.1" 200 143363 "-" "Masked UserAgent" - 0.095 0.071
    2.2.2.2 - - [09/Jul/2020:04:05:06 +0800] "GET /masked/request/uri/2 HTTP/1.1" 200 143388 "-" "Masked UserAgent 2" - 0.095 0.072

    返回write .. records on partition .. completed时,表示成功写入MaxCompute。写入成功

  3. 通过MaxCompute客户端或其他可以运行MaxCompute SQL的工具,执行如下命令,查询数据写入结果。

    set odps.sql.allow.fullscan=true;
    select * from logstash_test_groknginx;

    返回结果如下:

    +--------+------------+-------------+------------+------------+------------+------------+------------+------------+------------+------------+------------+
    | clientip   | remote_user | time       | verb       | uri        | version    | response   | body_bytes | referrer   | agent      | pt         |
    +------------+-------------+------------+------------+------------+------------+------------+------------+------------+------------+------------+
    | 1.1.1.1    | -           | 2020-07-09 01:02:03 | GET        | /masked/request/uri/1 | 1.1        | 200        | 0          | "-"        | "Masked UserAgent" | 2020-02-10       |
    | 2.2.2.2    | -           | 2020-07-09 04:05:06 | GET        | /masked/request/uri/2 | 1.1        | 200        | 0          | "-"        | "Masked UserAgent 2" | 2020-02-10       |
    +------------+-------------+------------+------------+------------+------------+------------+------------+------------+------------+------------+
    2 records (at most 10000 supported) fetched by instance tunnel.