使用 Flink 消费 CLS 日志

mysql -h 172.16.1.1 -uroot -p

create database if not exists flink_nginx;
use flink_nginx;
create table if not exists mysql_dest(
 ts timestamp,
 pv bigint,
 uv bigint
);

# 解压缩 Flink 二进制包
tar -xf flink-1.14.5-bin-scala_2.11.tgz
cd flink-1.14.5

# 下载 kafka 相关依赖
wget https://repo1.maven.org/maven2/org/apache/flink/flink-connector-kafka_2.11/1.14.5/flink-connector-kafka_2.11-1.14.5.jar
mv flink-connector-kafka_2.11-1.14.5.jar lib
wget https://repo1.maven.org/maven2/org/apache/kafka/kafka-clients/2.4.1/kafka-clients-2.4.1.jar
mv kafka-clients-2.4.1.jar lib

# 下载 MySQL 相关依赖
wget https://repo1.maven.org/maven2/org/apache/flink/flink-connector-jdbc_2.11/1.14.5/flink-connector-jdbc_2.11-1.14.5.jar
mv flink-connector-jdbc_2.11-1.14.5.jar lib
wget https://repo1.maven.org/maven2/mysql/mysql-connector-java/8.0.11/mysql-connector-java-8.0.11.jar
mv mysql-connector-java-8.0.11.jar lib
wget https://repo1.maven.org/maven2/org/apache/flink/flink-table-common/1.14.5/flink-table-common-1.14.5.jar
mv flink-table-common-1.14.5.jar lib

# 启动 Flink
bin/start-cluster.sh
bin/sql-client.sh

-- 消费CLS日志主题数据 ,建数据源表
CREATE TABLE `nginx_source`
(
    `remote_user` STRING,           -- 日志中字段，客户端名称
    `time_local` STRING,            -- 日志中字段，服务器本地时间
    `body_bytes_sent` BIGINT,       -- 日志中字段，发送给客户端的字节数
    `http_x_forwarded_for` STRING,  -- 日志中字段，当前端有代理服务器时，记录客户端真实 IP 地址的配置
    `remote_addr` STRING,           -- 日志中字段，客户端 IP 地址
    `protocol` STRING,              -- 日志中字段，协议类型
    `status` INT,                   -- 日志中字段，HTTP 请求状态码
    `url` STRING,                   -- 日志中字段，URL 地址
    `http_referer` STRING,          -- 日志中字段，该网页是从哪个页面链接过来的
    `http_user_agent` STRING,       -- 日志中字段，客户端浏览器信息
    `method` STRING,                -- 日志中字段，HTTP 请求方法
    `partition_id` BIGINT METADATA FROM 'partition' VIRTUAL,    -- CLS日志主题分区
    `ts` AS PROCTIME()                
)  WITH (
  'connector' = 'kafka',--将CLS日志主题当作Kafka Topic来消费
  'topic' = 'YourTopic',  -- CLS Kafka协议消费控制台给出的主题名称，例如12345-633a268c-XXXX-4a4c-XXXX-7a9a1a7baXXXX 
  'properties.bootstrap.servers' = 'kafkaconsumer-ap-guangzhou.cls.tencentcs.com:9096',   -- CLS Kafka协议消费控制台给出的服务地址，例子中是广州地域的外网消费地址，请按照您的实际情况填写
  'properties.group.id' = 'kafka_flink', -- Kafka 消费组名称
  '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";',--用户名是日志主题所属的日志集合ID，例如ca5cXXXX-dd2e-4ac0-af12-92d4b677d2c6，密码是用户的secretid#secrectkey组合的字符串，比AKIDWrwkHYYHjvqhz1mHVS8YhXXXX#XXXXuXtymIXT0Lac注意不要丢失#。建议使用子账号密钥，主账号为子账号授权时,遵循最小权限原则,即子账号的访问策略中的action、resource都配置为最小范围,可以满足操作即可.
  'properties.security.protocol' = 'SASL_PLAINTEXT',
  'properties.sasl.mechanism' = 'PLAIN'
);

--- 建立目标表
CREATE TABLE `mysql_dest`
(
    `ts` TIMESTAMP,  
    `pv` BIGINT, 
    `uv` BIGINT 
)  WITH (
    'connector' = 'jdbc',
    'url' = 'jdbc:mysql://11.150.2.1:3306/flink_nginx', -- 注意这边的时区设置
    'username'= 'username',     -- MYSQL账号
    'password'= 'password',     -- MYSQL密码
    'table-name' = 'mysql_dest' -- MYSQL表名
);

--- 查询 CLS日志主题，对Nginx日志中的PV，UV进行聚合计算，结果写入 MYSQL目标表
INSERT INTO mysql_dest (ts,uv,pv)
SELECT TUMBLE_START(ts, INTERVAL '1' MINUTE) start_ts, COUNT(DISTINCT remote_addr) uv,count(*) pv
FROM nginx_source
GROUP BY TUMBLE(ts, INTERVAL '1' MINUTE);