Synchronize MaxCompute data to Elasticsearch with Logstash

Configure a Logstash pipeline

1. Go to the Logstash Clusters page.

2. Navigate to the target cluster.

   1. In the top navigation bar, select the region where the cluster resides.

   2. On the Logstash Clusters page, find the cluster and click its ID.

3. In the left-side navigation pane, click Pipeline Management.

4. Click Create Pipeline.

5. On the Create Pipeline Task page, enter a pipeline ID and configure the pipeline.

   The following code provides an example configuration.

   input {
       maxcompute {
           access_id => "yourAccessKeyID"
           access_key => "yourAccessKeySecret"
           endpoint => "http://service.cn-hangzhou.maxcompute.aliyun-inc.com/api"
           project_name => "your_project_name"
           table_name => "your_table_name"
           partition => "sale_date='201911', region='hangzhou'"
           thread_num => 1
           dirty_data_file => "/ssd/1/<your_logstash_instance_id>/logstash/data/failed_records.txt"
       }
   }
   
   output {
     elasticsearch {
       hosts => ["http://es-cn-abcde12345.elasticsearch.aliyuncs.com:9200"]
       index => "your_index_name"
       user => "elastic"
       password => "your_password"
     }
   }

   Table 1. Input parameters

   Parameter	Type	Required	Description
   access_id	string	Yes	Your Alibaba Cloud AccessKey ID.
   access_key	string	Yes	Your Alibaba Cloud AccessKey Secret.
   endpoint	string	Yes	The MaxCompute endpoint in your region. For more information, see VPC Endpoint.
   project_name	string	Yes	The name of your MaxCompute project.
   table_name	string	Yes	The name of your MaxCompute table.
   partition	string	Yes	The partition key. For a partitioned table, you must specify the partition. For example: sale_date='201911', region='hangzhou'.
   thread_num	number	Yes	The number of threads. The default value is 1.
   dirty_data_file	string	Yes	The path to a file for logging failed records. Note The file must be located in the /ssd/1/<your_logstash_instance_id>/logstash/data/ directory.

   Important

   • The preceding configuration is for testing only. For production environments, configure the pipeline based on your business requirements. For other configuration options supported by the input plugin, see the logstash-input-maxcompute plugin documentation.

   • The logstash-input-maxcompute plugin performs a full data synchronization from MaxCompute to Alibaba Cloud Elasticsearch.

   For more information about pipeline configuration, see Logstash configuration files.

6. Click Next step and configure the pipeline parameters.

   Parameter	Description
   Pipeline Workers	Number of worker threads for the filter and output stages. Increase this value if events are backlogged or CPU is underutilized. Default: the number of CPU cores.
   Pipeline Batch Size	Maximum events a worker collects before executing filters and outputs. Larger batches increase memory usage and may require a larger JVM heap size (LS_HEAP_SIZE). Default: 125.
   Pipeline Batch Delay	Wait time in milliseconds before dispatching an undersized batch to a worker thread. Default: 50 ms.
   Queue Type	Internal queuing model for event buffering between stages. Valid values: MEMORY: Default. Uses an in-memory queue. PERSISTED: A disk-based persistent queue.
   Queue Max Bytes	The maximum amount of data the queue can store, in MB. The value must be an integer from 1 to 253-1. Default value: 1024. Note Make sure that this value is less than your total disk capacity.
   Queue Checkpoint Writes	Maximum events written before a checkpoint is forced (persistent queue only). 0 means no limit. Default: 1024.

   Warning

   Saving and deploying the configuration triggers an instance restart. Proceed only when this will not affect your business.

7. Click Save or Save and Deploy.

   • Save: Saves the pipeline configuration but does not apply it. After saving, you are returned to the Pipelines page. In the Pipelines section, you can click Deploy Now in the Actions column to restart the instance and apply the configuration.

   • Save and Deploy: Saves and deploys the configuration, restarting the instance to apply the changes.

Verify the results

1. Log on to the Kibana console of your destination Alibaba Cloud Elasticsearch instance.

   For more information, see Log on to the Kibana console.

2. In the left-side navigation pane, click Dev Tools.

3. In the Console, run the following command to view the synchronized data.

   GET /your_index_name/_search

   A successful response is shown below.

   
   {
     "took" : 6,
     "timed_out" : false,
     "_shards" : {
       "total" : 5,
       "successful" : 5,
       "skipped" : 0,
       "failed" : 0
     },
     "hits" : {
       "total" : 2,
       "max_score" : 1.0,
       "hits" : [
         {
           "_index" : "odps_index",
           "_type" : "doc",
           "_id" : "AT2dDxxxglu",
           "_score" : 1.0,
           "_source" : {
             "shop_name" : "shopy",
             "customer_id" : "y_id",
             "@timestamp" : "2019-12-16T07:30:15.208Z",
             "@version" : "1",
             "total_price" : 200.0
           }
         },
         {
           "_index" : "odps_index",
           "_type" : "doc",
           "_id" : "vtCdDxxxIT",
           "_score" : 1.0,
           "_source" : {
             "shop_name" : "shopy",
             "customer_id" : "y_id",
             "@timestamp" : "2019-12-16T07:30:15.206Z",
             "@version" : "1",
             "total_price" : 200.0
           }
         }
       ]
     }
   }
   

   Note

   If the command fails, check the logs on the Log Query page for troubleshooting. For more information, see Query Logs.

4. Go to the Monitoring page and click Indices.

5. On the Indices page, view the synchronized index and the number of documents it contains.

   The odps_index index status is Green with a Document Count of 4.

FAQ

Troubleshoot data write issues in Logstash