日志消费 CLS

CREATE TABLE `cls_source` (
  -- 需要自己替换成CLS日志中相应的字段
  `@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组合的字符串，比如AKIDWrwkHYYHjvqhz1mHVS8YhXXXX#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'
);

CREATE TABLE `cls_sink` (
    -- CLS对应日志主题的字段
    `logNo` STRING,
    `city` STRING,
    `__CONTENT__` STRING,
    `__PKG_LOGID__` STRING
) WITH (
    'connector' = 'kafka',
    -- CLS日志主题基本信息中的日志主题ID
    'topic' = 'CLS日志主题ID',
    'format' = 'json',
    'json.fail-on-missing-field' = 'false',
    'json.ignore-parse-errors' = 'true' ,
    -- CLS日志主题基本信息中通过Kafka协议上报数据的域名
    'properties.bootstrap.servers' = '${region}.cls.tencentyun.com:9095',
    -- 用户名是日志集合ID，例如ca5cXXXXdd2e-4ac0af12-92d4b677d2c6
    -- 密码是用户的SecretId#SecretKey组合的字符串，比如AKIDWrwkHYYHjvqhz1mHVS8YhXXXX#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'
);