Перейти к содержимому
Последнее обновление

AutoMQ

AutoMQ представляет собой облачную реализацию Kafka, переработанную специально для облачной среды. Благодаря переносу хранения данных в объектное хранилище AutoMQ сохраняет 100% совместимость с Apache Kafka, обеспечивая снижение затрат до 10 раз и повышение эластичности до 100 раз. Инновационная архитектура с общим хранилищем обеспечивает миграцию партиций за секунды, автоматическую балансировку трафика и автоматическое масштабирование за секунды, сохраняя высокую пропускную способность и низкую задержку.

В этом документе описано, как использовать Routine Load в Apache Doris для чтения данных JSON из совместимого с Kafka Topic в AutoMQ и их непрерывной загрузки в таблицу Doris.

Архитектура хранения AutoMQ

Сценарии использования​

Если бизнес-данные уже записаны в AutoMQ и вы хотите непрерывно анализировать их в Apache Doris в реальном времени, используйте Routine Load для чтения данных из Topic в AutoMQ. Пример в этом документе использует Topic с данными JSON и тестовую таблицу Doris, чтобы продемонстрировать весь процесс: от подготовки данных до проверки результата загрузки.

Общий порядок действий:

  1. Подготовьте тестовую базу данных и тестовую таблицу Apache Doris.
  2. Подготовьте кластер AutoMQ, инструменты командной строки Kafka и тестовый Topic.
  3. Запишите тестовые данные JSON в Topic в AutoMQ.
  4. Создайте задание загрузки Routine Load в Doris.
  5. Выполните запрос к таблице Doris, чтобы проверить успешную загрузку данных.

Предварительные требования​

Перед началом работы убедитесь, что следующая среда готова к использованию.

КомпонентОписание
Кластер Apache DorisУбедитесь, что кластер Apache Doris доступен. Для демонстрации в этом документе используется тестовая среда в Linux, развёрнутая по инструкции Быстрый старт.
Кластер AutoMQРазверните доступный кластер по официальной документации по развёртыванию AutoMQ и убедитесь, что между AutoMQ и Apache Doris есть сетевое соединение.
Инструменты командной строки KafkaСкачайте последний пакет TGZ со страницы релизов AutoMQ и распакуйте его. В этом документе предполагается, что каталог с распакованными файлами обозначен как $AUTOMQ_HOME, а для создания Topic и генерации тестовых данных используются команды инструментов из $AUTOMQ_HOME/bin.

Подготовка тестовой таблицы Doris​

Создайте базу данных и тестовую таблицу в Doris. Последующие примеры по умолчанию выполняются в automq_db.

CREATE DATABASE automq_db;
USE automq_db;

CREATE TABLE automq_db.users (
    id BIGINT NOT NULL,
    name STRING NOT NULL,
    timestamp STRING NULL,
    status STRING NULL
) DISTRIBUTED BY HASH(id) PROPERTIES ("replication_num" = "1");

Подготовка Topic AutoMQ и тестовых данных​

В этом документе используются следующие примеры значений параметров. При выполнении команд замените адрес из примера фактическим адресом AutoMQ Bootstrap Server.

ПараметрПример значенияОписание
AutoMQ Bootstrap Server127.0.0.1:9092Адрес для доступа к кластеру AutoMQ.
Topicexample_topicTopic для хранения тестовых данных JSON.
Каталог с распакованными файлами AutoMQ$AUTOMQ_HOMEКаталог, в который распакован пакет TGZ AutoMQ.

Создание Topic​

Используйте инструмент командной строки Apache Kafka для создания example_topic.

$AUTOMQ_HOME/bin/kafka-topics.sh \
    --create \
    --topic example_topic \
    --bootstrap-server 127.0.0.1:9092 \
    --partitions 1 \
    --replication-factor 1

После создания выполните следующую команду, чтобы убедиться, что Topic успешно создан.

$AUTOMQ_HOME/bin/kafka-topics.sh \
    --describe \
    --topic example_topic \
    --bootstrap-server 127.0.0.1:9092

Подготовка тестовых данных​

Тестовые данные используют формат JSON. Поля должны соответствовать таблице Doris и jsonpaths в задании Routine Load, которое будет создано далее.

{
    "id": 1,
    "name": "Test User",
    "timestamp": "2023-11-10T12:00:00",
    "status": "active"
}

Запись тестовых данных​

Запишите тестовые данные в example_topic с помощью инструмента командной строки Kafka или программным способом. В следующем примере для записи данных используется инструмент командной строки.

echo '{"id": 1, "name": "Test User", "timestamp": "2023-11-10T12:00:00", "status": "active"}' | $AUTOMQ_HOME/bin/kafka-console-producer.sh \
    --bootstrap-server 127.0.0.1:9092 \
    --topic example_topic

Выполните следующую команду, чтобы просмотреть данные, только что записанные в Topic.

$AUTOMQ_HOME/bin/kafka-console-consumer.sh \
    --bootstrap-server 127.0.0.1:9092 \
    --topic example_topic \
    --from-beginning

Создание задания загрузки Routine Load​

В командной строке Apache Doris создайте задание Routine Load для непрерывного чтения данных JSON из Topic в AutoMQ. Подробное описание параметров Routine Load приведено в разделе Doris Routine Load.

CREATE ROUTINE LOAD automq_example_load ON users
COLUMNS(id, name, timestamp, status)
PROPERTIES
(
    "format" = "json",
    "jsonpaths" = "[\"$.id\",\"$.name\",\"$.timestamp\",\"$.status\"]"
)
FROM KAFKA
(
    "kafka_broker_list" = "127.0.0.1:9092",
    "kafka_topic" = "example_topic",
    "property.kafka_default_offsets" = "OFFSET_BEGINNING"
);

Основные параметры описаны ниже.

ПараметрПример значенияОписание
formatjsonУказывает, что загружаемые данные имеют формат JSON.
jsonpaths["$.id","$.name","$.timestamp","$.status"]Задаёт соответствие между полями JSON и полями таблицы Doris.
kafka_broker_list127.0.0.1:9092Адрес AutoMQ Bootstrap Server. При выполнении замените его фактическим адресом.
kafka_topicexample_topicTopic в AutoMQ, из которого Routine Load должен читать данные.
property.kafka_default_offsetsOFFSET_BEGINNINGЧтение данных с начала Topic.

Проверка загрузки данных​

Сначала проверьте состояние задания загрузки Routine Load, чтобы убедиться, что оно выполняется.

SHOW ROUTINE LOAD\G;

Затем выполните запрос к таблице Doris, чтобы убедиться, что тестовые данные загружены.

SELECT * FROM users;
+------+-----------+---------------------+--------+
| id   | name      | timestamp           | status |
+------+-----------+---------------------+--------+
|    1 | Test User | 2023-11-10T12:00:00 | active |
+------+-----------+---------------------+--------+
1 row in set (0.01 sec)

Примечания​

  • При выполнении команд из примеров замените 127.0.0.1:9092 фактическим адресом AutoMQ Bootstrap Server.
  • Поля тестовых данных JSON должны соответствовать полям таблицы Doris и настройке jsonpaths в Routine Load.
  • Перед созданием задания Routine Load убедитесь, что Doris имеет доступ к кластеру AutoMQ.
Report issue