-- Table creation statements are as follows:
CREATE TABLE nginx_source (
  -- Fields in the logs.
  @metadata STRING,
  @timestamp TIMESTAMP,
  agent STRING,
  ecs STRING,
  host STRING,
  input STRING,
  log STRING,
  message STRING,
  partition_id BIGINT METADATA
  FROM
    'partition' VIRTUAL,
    -- Kafka partition.
    ts TIMESTAMP(3) METADATA
  FROM
    'timestamp'
) WITH (
  'connector' = 'kafka',
  -- The topic name provided by the CLS Kafka protocol consumption console, for example, XXXXXX-633a268c-XXXX-4a4c-XXXX-7a9a1a7baXXXX, can be copied from the console.
  'topic' = 'Your consumption topics',
  -- Service address + port, for example, public network port 9096 and private network port 9095, for private network consumption. Please enter according to your actual situation.
  'properties.bootstrap.servers' = 'kafkaconsumer-${region}.cls.tencentyun.com:9095',
  -- Please replace with your consumer group name.   
  properties.group.id' = 'Consumer group ID',
  'scan.startup.mode' = 'earliest-offset',
  'format' = 'json',
  'json.fail-on-missing-field' = 'false',
  'json.ignore-parse-errors' = 'true',
  -- The username is the logset ID, for example, ca5cXXXXdd2e-4ac0af12-92d4b677d2c6.
  -- The password is a string combining the user's SecretId#SecretKey, for example, AKIDWrwkHYYHjvqhz1mHVS8YhXXXX#XXXXuXtymIXT0Lac. Be careful not to lose the #. We recommend that you use the sub-account keys. When authorizing a sub-account, follow the principle of least privilege, configuring the action and resource in the sub-account access policy to the minimum range to fulfill the operations. Note that jaas.config ends with a semicolon. An error will be reported if not filled in.
  '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'
);