This document describes how to consume CLS logs with Flink in real time, analyze Nginx log data with Flink SQL, calculate web PVs/UVs, and write the result to self-built MySQL databases in real time.
Components/Applications and their versions used in this document are as described below:
2. Install CLS's log collector LogListener as instructed in LogListener Installation Guide. Similar to the open-source component Beats, LogListener is an agent that collects logs.
3. After index is enabled for log topics, you can query Nginx logs as shown below:
4. Enable consumption over Kafka in the CLS console to use the feature. You can consume a log topic as a Kafka topic. This document describes how to consume Nginx log data in real time with the stream computing framework Flink and then write the real-time computing result to MySQL.
3. The installation is successful if the following information is displayed. Note that the default web port is 8081.
Step 5. Consume CLS log data with Flink
1. On the SQL client UI, execute the following SQL statements:
-- Create a data source table to consume Kafka data
CREATE TABLE `nginx_source`
(
`remote_user` STRING, -- Field in the log, which indicates the client name.
`time_local` STRING, -- Field in the log, which indicates the localtime of the server.
`body_bytes_sent` BIGINT, -- Field in the log, which indicates the number of bytes sent to the client.
`http_x_forwarded_for` STRING, -- Field in the log, which records the actual client IP when there is a proxy server on the frontend.
`remote_addr` STRING, -- Field in the log, which indicates the client IP.
`protocol` STRING, -- Field in the log, which indicates the protocol type.
`status` INT, -- Field in the log, which indicates the HTTP request status code.
`url` STRING, -- Field in the log, which indicates the URL.
`http_referer` STRING, -- Field in the log, which indicates the URL of the referer.
`http_user_agent` STRING, -- Field in the log, which indicates the client browser information.
`method` STRING, -- Field in the log, which indicates the HTTP request method.
`partition_id` BIGINT METADATA FROM 'partition' VIRTUAL, -- Kafka partition
`ts` AS PROCTIME()
) WITH (
'connector'='kafka',
'topic'='YourTopic', -- Topic name provided in the CLS console for consumption over Kafka, such as `out-633a268c-XXXX-4a4c-XXXX-7a9a1a7baXXXX`
'properties.bootstrap.servers'='kafkaconsumer-ap-guangzhou.cls.tencentcs.com:9096', -- Service address provided in the CLS console for consumption over Kafka. The public network consumer address in Guangzhou region is used as an example. You need to enter the actual information.
'properties.group.id'='kafka_flink', -- Kafka consumer group name
'scan.startup.mode'='earliest-offset',
'format'='json',
'json.fail-on-missing-field'='false',
'json.ignore-parse-errors'='true' ,
'properties.sasl.jaas.config'='org.apache.kafka.common.security.plain.PlainLoginModule required username="your username" password="your password";',--Your username is the logset ID of the log topic, such as `ca5cXXXX-dd2e-4ac0-af12-92d4b677d2c6`, and the password is a string of your `secretid#secrectkey`, such as `AKIDWrwkHYYHjvqhz1mHVS8YhXXXX#XXXXuXtymIXT0Lac`. Note that `#` is required. We recommend that you use a sub-account key and follow the principle of least privilege when authorizing a sub-account, that is, configure the minimum permission for`action` and `resource`in the access policy of the sub-account.
'properties.security.protocol'='SASL_PLAINTEXT',
'properties.sasl.mechanism'='PLAIN'
);
--- Create the target table and write it to MySQL
CREATE TABLE `mysql_dest`
(
`ts` TIMESTAMP,
`pv` BIGINT,
`uv` BIGINT
) WITH (
'connector'='jdbc',
'url'='jdbc:mysql://11.150.2.1:3306/flink_nginx?&serverTimezone=Asia/Shanghai', -- Note the time zone settings here
'username'='username', -- MySQL account
'password'='password', -- MySQL password
'table-name'='mysql_dest' -- MySQL table name
);
--- Query the Kafka data source table and write the computing result to the MySQL target table
