-- CLSログトピックデータを消費し、データソーステーブルを作成します。
CREATE TABLE `nginx_source`
(
    `}
    `time_local` STRING,            -- ログ内フィールド、サーバーローカル時間
    `}
    `}
    `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を組み合わせた文字列です。例えばAKID********************************#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);