request_logs system table

Schema

CREATE TABLE `request_logs` (
  `version` STRING COMMENT 'Version of the catalog event',
  `event` STRING COMMENT 'Enumeration value',
  `detail` text COMMENT 'Detailed data in JSON format',
  `identifier` STRING COMMENT 'Identifier for this event',
  `requested_by` STRING COMMENT 'Requesting user',
  `requested_at` BIGINT COMMENT 'Request time in milliseconds',
  `dt` STRING COMMENT 'Date partition, for example, 20250505'
) PARTITIONED BY (dt) WITH (
   'file.format' = 'avro',
   'partition.expiration-time' = '30 d',
   'partition.timestamp-formatter' = 'yyyyMMdd'
);

Partitions expire after 30 days.

Event types

The event field is an enumeration. The identifier and detail fields vary by event type.

All xxxRequest class definitions are from the org.apache.paimon.rest.requests package.

Prerequisites

Before you begin, ensure that you have:

• A Realtime Compute for Apache Flink instance

• A Paimon DLF catalog

Run a batch query

On the Data Exploration page, run the following SQL query, replacing <Your_Flink_Catalog_Name> with your Flink catalog name and the dt value with the target date:

SELECT * FROM `<Your_Flink_Catalog_Name>`.`system`.`request_logs` WHERE dt='20250601';

Prerequisites

Before you begin, ensure that you have:

• A Realtime Compute for Apache Flink instance

• A Paimon DLF catalog

• An ApsaraMQ for Kafka instance with a topic created and deployed

Set up the Flink stream job

On the ETL page, create a Flink stream job with the following SQL, then deploy and start the job:

CREATE TEMPORARY TABLE `kafka_sink` (
  version STRING,
  event STRING,
  detail STRING,
  identifier STRING,
  requested_by STRING,
  requested_at BIGINT,
  dt STRING
) WITH (
  'connector' = 'kafka',
  'topic' = '<your-topic-name>',
  'properties.bootstrap.servers' = '<your-kafka-endpoint>',
  'properties.enable.idempotence'='false',
  'format' = 'json'
);

INSERT INTO `kafka_sink`
SELECT *
FROM `flink_test`.`system`.`request_logs`/*+ OPTIONS('scan.mode' = 'latest') */;

The kafka_sink table parameters:

Consume and parse log messages

The following Java examples show how to consume messages from Kafka and parse the log data.

Step 1: Define the RequestLog class

Map each field in the request_logs schema to a Java class using Jackson @JsonProperty annotations:

import com.fasterxml.jackson.annotation.JsonProperty;

public class RequestLog {

    @JsonProperty("version")
    private String version;

    @JsonProperty("event")
    private String event;

    @JsonProperty("detail")
    private String detail;

    @JsonProperty("identifier")
    private String identifier;

    @JsonProperty("requested_by")
    private String requestedBy;

    @JsonProperty("requested_at")
    private long requestedAt;

    @JsonProperty("dt")
    private String dt;

    public RequestLog() {}

    public String getVersion() { return version; }
    public void setVersion(String version) { this.version = version; }

    public String getEvent() { return event; }
    public void setEvent(String event) { this.event = event; }

    public String getDetail() { return detail; }
    public void setDetail(String detail) { this.detail = detail; }

    public String getIdentifier() { return identifier; }
    public void setIdentifier(String identifier) { this.identifier = identifier; }

    public String getRequestedBy() { return requestedBy; }
    public void setRequestedBy(String requestedBy) { this.requestedBy = requestedBy; }

    public long getRequestedAt() { return requestedAt; }
    public void setRequestedAt(long requestedAt) { this.requestedAt = requestedAt; }

    public String getDt() { return dt; }
    public void setDt(String dt) { this.dt = dt; }
}

Step 2: Parse the detail field

Use RequestLogParser to map each event type to its corresponding request class and deserialize the detail JSON field using RESTApi.fromJson(). For Maven dependency configuration, see the API guide.

package com.aliyun.morax.flink;

import org.apache.paimon.rest.RESTApi;
import org.apache.paimon.rest.requests.CreateTableRequest;
import org.apache.paimon.rest.requests.RenameTableRequest;

import java.util.HashMap;
import java.util.Map;

public class RequestLogParser {

    // Map event types to their corresponding request classes.
    private static final Map<String, Class<?>> EVENT_CLASS_MAP = new HashMap<>();

    static {
        EVENT_CLASS_MAP.put("CreateTable", CreateTableRequest.class);
        EVENT_CLASS_MAP.put("RenameTable", RenameTableRequest.class);
        EVENT_CLASS_MAP.put("DropTable", DropTableRequest.class);
        EVENT_CLASS_MAP.put("CreateDatabase", CreateDatabaseRequest.class);
        EVENT_CLASS_MAP.put("DropDatabase", DropDatabaseRequest.class);
        EVENT_CLASS_MAP.put("AlterTable", AlterTableRequest.class);
        // Add more event types as needed.
    }

    public Object parseDetail(RequestLog log) {
        Class<?> clazz = EVENT_CLASS_MAP.get(log.getEvent());
        if (clazz == null) {
            throw new UnsupportedOperationException("Unsupported event: " + log.getEvent());
        }
        return fromJson(log.getDetail(), clazz);
    }

    private <T> T fromJson(String json, Class<T> clazz) {
        try {
            return RESTApi.fromJson(json, clazz);
        } catch (Exception e) {
            throw new RuntimeException("Failed to parse JSON for class: " + clazz.getSimpleName(), e);
        }
    }
}

Step 3: Create a Kafka consumer

The following example shows a complete Kafka consumer that reads log records and parses each entry. For detailed configuration and additional parameter descriptions, see Use instance endpoints to send and receive messages.

package com.aliyun.morax.flink;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.paimon.rest.RESTApi;
import org.apache.paimon.rest.requests.CreateTableRequest;
import org.apache.paimon.rest.requests.RenameTableRequest;
import org.apache.paimon.shade.com.fasterxml.jackson.databind.ObjectMapper;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

public class KafkaConsumerWithRequestLogParser {

    private static final ObjectMapper objectMapper = new ObjectMapper();
    private static final RequestLogParser parser = new RequestLogParser();

    public static void main(String[] args) {
        Properties kafkaProperties = JavaKafkaConfigurer.getKafkaProperties();

        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getProperty("bootstrap.servers"));
        // Add other consumer properties as needed.

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

        List<String> topics = new ArrayList<>();
        String topicStr = kafkaProperties.getProperty("topic");
        for (String topic : topicStr.split(",")) {
            topics.add(topic.trim());
        }
        consumer.subscribe(topics);

        while (true) {
            try {
                ConsumerRecords<String, String> records = consumer.poll(java.time.Duration.ofSeconds(1));
                for (ConsumerRecord<String, String> record : records) {
                    try {
                        // Deserialize the record into a RequestLog object.
                        RequestLog log = objectMapper.readValue(record.value(), RequestLog.class);
                        // Parse the detail field based on the event type.
                        Object detail = parser.parseDetail(log);

                        // Process the parsed detail as needed.
                    } catch (Exception e) {
                        System.err.println("Error processing record: " + record.offset());
                        e.printStackTrace();
                    }
                }
            } catch (Exception e) {
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException ignore) {
                    Thread.currentThread().interrupt();
                }
                System.err.println("Consumer error, retrying...");
                e.printStackTrace();
            }
        }
    }
}