阿里云云原生大数据计算服务 MaxCompute使用Logstash（流式数据传输）-云淘科技

前提条件

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

• 已安装Logstash并创建Logstash日志收集实例。

  更多信息，请参见Getting Started with Logstash。

• 已创建目标MaxCompute项目。

  更多创建MaxCompute项目信息，请参见创建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 => ""
                aliyun_access_key => ""
                aliyun_mc_endpoint => ""
                project => ""
                table => ""
                partition => "pt=$"
                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


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


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


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


如果某个分区的值为解析后的日志中一个日期时间字段的值，并且需要进行重新格式化，格式为{分区列名}=$。其中：{时间格式}是重新格式化的格式化字符串。
在本示例中，将格式化到仅保留日期（%F）。如果要按照日期date作为第一级分区，小时hour作为第二级分区，配置格式为"date=$,hour=$"。


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






partition_time_format


可选。指定当一个字符串型的日期时间字段被分区信息引用时，该字段的源格式字符串。
在本例中，时间字段timestamp已经被date插件转换为时间类型，因此不需指定。
即使未使用date过滤插件进行转换，亦未指定此配置项的值，在大多数情况下插件仍然可以自动识别内容为日期时间的字符串，并自动完成需要的转换。即只在少数自动识别失败的情况下需要手动指定此项的值。
如果不使用date过滤插件，而是手动进行转换，则需要配置如下信息：


手动指定partition_time_format：partition_time_format => "%d/%b/%Y:%H:%M:%S %z"。


将分区中引用的字段改为日志中的字符串字段：partition => "pt=$"。






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文档。