How to use the Logstash output plugin logstash-output-maxcompute-MaxCompute(MaxCompute)-阿里云帮助中心

MaxCompute can ingest logs collected by open-source Logstash. Use the logstash-output-maxcompute output plugin to upload data to MaxCompute through the Streaming Tunnel.

Prerequisites

Complete the following prerequisites:

You have installed Logstash for log collection. For more information, see Getting Started with Logstash.
You have created a MaxCompute project.

Background information

Logstash is an open-source, server-side data processing pipeline that can ingest and transform data from multiple data sources simultaneously, and then send it to a destination data store. Use the Logstash logstash-output-maxcompute output plugin to upload log data to MaxCompute using the Streaming Tunnel.

The logstash-output-maxcompute plugin is based on Logstash v7.8.0 and can be used as an output. The plugin provides the following features:

It uses the Streaming Tunnel to prevent concurrency issues and the creation of small files common when importing data with the Batch Data Tunnel.
The plugin can generate partition fields based on parsed log fields and automatically create partitions that do not exist.

The logstash-output-maxcompute plugin is suitable for the following use cases:

The log format of your application is supported by a Logstash input plugin or is easy to parse, such as NGINX logs.
You want to automatically create partitions and import data based on log content.

The logstash-output-maxcompute plugin supports the following data types: STRING, BIGINT, DOUBLE, DATETIME, and BOOLEAN.

Note

The format of DATETIME fields in logs is automatically inferred using the ruby Time.parse function.
If a boolean field in a log satisfies the .to_string().lowercase() == "true" condition, it evaluates to true; otherwise, it evaluates to false.

This topic uses NGINX logs as an example to explain how to configure and use the plugin.

Step 1: Download and install the plugin

You can download a Logstash instance with the logstash-output-maxcompute plugin pre-installed from this Logstash instance download link. If you use this method, you can skip the installation and proceed to the next step. To install the plugin manually:

Download the logstash-output-maxcompute plugin from the logstash-output-maxcompute plugin download link and place it in the %logstash% root directory of Logstash.
In the %logstash% root directory, open the Gemfile configuration file and replace source "https://rubygems.org" with source 'https://gems.ruby-china.com'.

On a Windows system, open the command-line interface (CLI), navigate to the %logstash% root directory, and run the following command to install the logstash-output-maxcompute output plugin.

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

The Installation successful message confirms the installation.

D:\logstash>
D:\logstash>bin\logstash-plugin install logstash-output-maxcompute-1.1.0.gem
Validating logstash-output-maxcompute-1.1.0.gem
Installing logstash-output-maxcompute
Installation successful

Optional: Run the following command to verify the installation.
```
bin\logstash-plugin list maxcompute
```
Note
On a Linux system, run the command bin/logstash-plugin list maxcompute.
A successful installation returns logstash-output-maxcompute. If the installation fails, see RubyGems for solutions.
```
D:\logstash>bin\logstash-plugin list maxcompute
logstash-output-maxcompute
```

Step 2: Create a destination table

Use the MaxCompute client or another SQL-compatible tool to run the following command to create a destination table, such as logstash_test_groknginx, in your MaxCompute project. Log data will be imported into this table based on date partitions.

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);

Step 3: Create a pipeline configuration file

In the %logstash% root directory of Logstash, create a configuration file named pipeline.conf and add the following content:

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", "body_bytes", "referrer", "agent"]
        }
}

Parameter	Description
your_accesskey_id	An AccessKey ID with permissions to access the destination MaxCompute project.
your_accesskey_secret	The AccessKey Secret that corresponds to the AccessKey ID.
your_project_endpoint	The region endpoint for your MaxCompute project. For more information about endpoints, see Endpoints and headers.
your_project_name	The name of the destination MaxCompute project.
table_name	The name of the destination table that you created in Step 2.
partition	Specifies how the plugin generates partition information from log fields. If the destination table has multiple partition levels, you must specify values for all partition levels. The following formats are supported: For a constant partition value, use the format: `{partition_column_name}={constant_value}`. If a partition value comes from a field in the parsed log, use the format: `{partition_column_name}=$<{log_field_name}>`. If a partition value comes from a datetime field in the parsed log and requires reformatting, use the format: `{partition_column_name}=$<{log_field_name}.strftime('{time_format}')>`. In this format, `{time_format}` is the new format string. In this example, the timestamp is formatted to retain only the date (%F). If you want to use the date `date` as the first-level partition and the hour `hour` as the second-level partition, the configuration format is `"date=$<timestamp.strftime('%F')>,hour=$<timestamp.strftime('%H')>"`. Use commas (,) to separate multiple partition levels. The order of the partitions must match the order specified in the CREATE TABLE statement.
partition_time_format	Optional. Specifies the source format string for a string-type datetime field when it is referenced in the partition information. In this example, the time field `timestamp` has already been converted to the time data type by the `date` plugin, so you do not need to specify it. Even if you do not use the `date` filter plugin for conversion and do not specify a value for this configuration option, in most cases, the plugin can still automatically recognize strings as date and time values and perform the necessary conversion. This means you only need to manually specify a value for this option in the few cases where automatic recognition fails. If you do not use the `date` filter plugin and instead perform the conversion manually, you need to configure the following information: Manually specify partition_time_format: `partition_time_format => "%d/%b/%Y:%H:%M:%S %z"`. Change the field referenced by the partition to a string field from the log: `partition => "pt=$<httptimestamp.strftime('%F')>"`.
value_fields	Specifies the log fields that map to the fields in the destination table. The order of the log fields must match the order of the table fields. The fields in the destination table are ordered as `clientip string, remote_user string, time datetime, verb string, uri string, version string, response string, body_bytes bigint, referrer string, agent string`, which correspond to `"clientip", "remote_user", "timestamp", "verb", "request", "httpversion", "response", "bytes", "referrer", "agent"` respectively.
aliyun_mc_tunnel_endpoint	Optional. You can use this parameter to specify a Tunnel endpoint, which overrides the automatic routing mechanism.
retry_time	The number of retries after a failed write operation. Default value: 3.
retry_interval	The minimum retry interval, in seconds. Default value: 1.
batch_size	The maximum number of log entries to process in a single batch. Default value: 100.
batch_timeout	The timeout period for writing data to MaxCompute, in seconds. Default value: 5.

Note

In this configuration file, the specified log input is standard input(input { stdin {} }). In a real-world scenario, you can use the Logstash File input plugin to automatically read NGINX logs from a local hard drive. For more information, see the Logstash documentation.

Step 4: Run and test

On a Windows system, open the CLI, navigate to the %logstash% root directory, and run the following command to start Logstash.

bin\logstash -f pipeline.conf

After the command is successfully executed, the returned message Successfully started Logstash API endpoint indicates that Logstash has started successfully and the API port is 9600.

D:\logstash>bin\logstash -f pipeline.conf
Sending Logstash logs to D:/logstash/logs which is now configured via log4j2.properties
[2021-01-27T17:46:34,347][WARN ][logstash.config.source.multilocal] Ignoring the 'pipelines.yml' file because modules or command line options are specified
[2021-01-27T17:46:34,679][INFO ][logstash.runner          ] Starting Logstash {"logstash.version"=>"7.8.0", "jruby.version"=>"jruby 9.2.11.1 (2.5.7) 2020-03-25 b1f55b1a40 Java HotSpot(TM) 64-Bit Server VM 25.131-b11 on 1.8.0_131-b11 +indy +jit [mswin32-x86_64]"}
[2021-01-27T17:46:40,972][INFO ][org.reflections.Reflections] Reflections took 241 ms to scan 1 urls, producing 21 keys and 41 values
[2021-01-27T17:46:44,030][INFO ][logstash.javapipeline    ][main] Starting pipeline {:pipeline_id=>"main", "pipeline.workers"=>4, "pipeline.batch.size"=>125, "pipeline.batch.delay"=>50, "pipeline.max_inflight"=>500, "pipeline.sources"=>["D:/logstash/pipeline.conf"], :thread=>"#&lt;Thread:0x37536680 run&gt;"}
[2021-01-27T17:46:46,947][INFO ][logstash.javapipeline    ][main] Pipeline started {"pipeline.id"=>"main"}
The stdin plugin is now waiting for input:
[2021-01-27T17:46:47,174][INFO ][logstash.agent           ] Pipelines running {:count=>1, :running_pipelines=>[:main], :non_running_pipelines=>[]}
[2021-01-27T17:46:48,185][INFO ][logstash.agent           ] Successfully started Logstash API endpoint {:port=>9600}

In the CLI, paste the following sample logs and press 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

After a successful execution, the terminal returns write .. records on partition .. completed. This indicates that Logstash has started successfully and written the Nginx logs to the MaxCompute table.

[2021-01-27T18:00:47,196][INFO ][logstash.javapipeline    ][main] Pipeline started {"pipeline.id"=>"main"}
The stdin plugin is now waiting for input:
[2021-01-27T18:00:47,388][INFO ][logstash.agent           ] Pipelines running {:count=>1, :running_pipelines=>[:main], :non_running_pipelines=>[]}
[2021-01-27T18:00:48,323][INFO ][logstash.agent           ] Successfully started Logstash API endpoint {:port=>9600}
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
[2021-01-27T18:00:56,757][INFO ][logstash.outputs.maxtunnel][main][691d6be762308ae955883c0cf0719e0a71754924lbe162d0d309f79690c9e2448] write 1 records on table doc_test_dev.logstash_test_groknginx partition pt='2020-07-08' completed. TraceId: 20210127180041e230f60b00012894
[2021-01-27T18:01:00,093][INFO ][logstash.outputs.maxtunnel][main] write 1 records on table doc_test_dev.logstash_test_groknginx partition pt='2020-07-08' completed. TraceId: 20210127180044b31f60b0001369e

Use the MaxCompute client or another SQL-compatible tool to run the following command to query the data and verify the result.

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

The following output is returned:

+--------+------------+-------------+------------+------------+------------+------------+------------+------------+------------+------------+------------+
| 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-07-08       |
| 2.2.2.2    | -           | 2020-07-09 04:05:06 | GET        | /masked/request/uri/2 | 1.1        | 200        | 0          | "-"        | "Masked UserAgent 2" | 2020-07-08       |
+------------+-------------+------------+------------+------------+------------+------------+------------+------------+------------+------------+
2 records (at most 10000 supported) fetched by instance tunnel.