CREATE TABLE `nginx_source`( # ログ内のフィールド`@metadata` STRING,`@timestamp` TIMESTAMP,`agent` STRING,`ecs` STRING,`host` STRING,`input` STRING,`log` STRING,`message` STRING,`partition_id` BIGINT METADATA FROM 'partition' VIRTUAL, -- kafkaパーティション`ts` TIMESTAMP(3) METADATA FROM 'timestamp') WITH ('connector' = 'kafka',# cls kafkaプロトコルでコンソールから提供されるトピック名です。例えばXXXXXX-633a268c-XXXX-4a4c-XXXX-7a9a1a7baXXXXは、コンソールでコピーできます。'topic' = 'お客様の消費トピック',# サービスアドレス+ポート、外部ネットワークポートは9096、内部ネットワークポートは9095です。例は内部ネットワーク消費です。お客様の実際の状況に応じて記入してください。'properties.bootstrap.servers' = 'kafkaconsumer-${region}.cls.tencentyun.com:9095',# お客様のコンシューマーグループ名に置き換えてください。'properties.group.id' = 'お客様のコンシューマーグループ名','scan.startup.mode' = 'earliest-offset','format' = 'json','json.fail-on-missing-field' = 'false','json.ignore-parse-errors' = 'true' ,# ユーザー名はログセットIDです。例えばca5cXXXXdd2e-4ac0af12-92d4b677d2c6です。# パスワードはユーザーのSecretId#SecretKeyを組み合わせた文字列です。例えばAKID********************************#XXXXuXtymIXT0Lacです。#を失わないように注意してください。サブアカウントキーの使用を推奨します。メインアカウントがサブアカウントに権限を付与する際は、最小権限の原則に従ってください。つまり、サブアカウントのアクセスポリシーにおけるaction、resourceはすべて最小範囲に設定し、操作を満たせば十分です。jaas.configの最後には;セミコロンが必要です。記入しないとエラーが発生します。'properties.sasl.jaas.config' = 'org.apache.kafka.common.security.plain.PlainLoginModule required username="${logsetID}" password="${SecretId}#${SecretKey}";','properties.security.protocol' = 'SASL_PLAINTEXT','properties.sasl.mechanism' = 'PLAIN');
<dependency><groupId>org.apache.flink</groupId><artifactId>flink-connector-kafka</artifactId><version>1.14.4</version></dependency>
CREATE TABLE `nginx_source`(# ログ内のフィールド`@metadata` STRING,`@timestamp` TIMESTAMP,`agent` STRING,`ecs` STRING,`host` STRING,`input` STRING,`log` STRING,`message` STRING,# kafkaパーティション`partition_id` BIGINT METADATA FROM 'partition' VIRTUAL,`ts` TIMESTAMP(3) METADATA FROM 'timestamp') WITH ('connector' = 'kafka',# cls kafkaプロトコルでコンソールから提供されるトピック名です。例えばXXXXXX-633a268c-XXXX-4a4c-XXXX-7a9a1a7baXXXXは、コンソールでコピーできます。'topic' = 'お客様の消費トピック',# サービスアドレス+ポート、外部ネットワークポートは9096、内部ネットワークポートは9095です。例は内部ネットワーク消費です。お客様の実際の状況に応じて記入してください。'properties.bootstrap.servers' = 'kafkaconsumer-${region}.cls.tencentyun.com:9095',# お客様のコンシューマーグループ名に置き換えてください。'properties.group.id' = 'お客様のコンシューマーグループ名','scan.startup.mode' = 'earliest-offset','format' = 'json','json.fail-on-missing-field' = 'false','json.ignore-parse-errors' = 'true' ,# ユーザー名はログセットIDです。例えばca5cXXXXdd2e-4ac0af12-92d4b677d2c6です。# パスワードはユーザーのSecretId#SecretKeyを組み合わせた文字列です。例えばAKID********************************#XXXXuXtymIXT0Lacです。#を失わないように注意してください。サブアカウントキーの使用を推奨します。メインアカウントがサブアカウントに権限を付与する際は、最小権限の原則に従ってください。つまり、サブアカウントのアクセスポリシーにおけるaction、resourceはすべて最小範囲に設定し、操作を満たせば十分です。jaas.configの最後には;セミコロンが必要です。記入しないとエラーが発生します。'properties.sasl.jaas.config' = 'org.apache.kafka.common.security.plain.PlainLoginModule required username="${logsetID}" password="${SecretId}#${SecretKey}";','properties.security.protocol' = 'SASL_PLAINTEXT','properties.sasl.mechanism' = 'PLAIN');
select count(*) , host from nginx_source group by host;
フィードバック