Logstash日志数据流式写入MaxCompute

步骤一：下载并安装插件

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

1. 下载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信息。

步骤二：创建目标表

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配置文件

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"]
        }
}

步骤四：运行和测试

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的工具，执行如下命令，查询数据写入结果。

   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.