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

Быстрый старт

На этой странице приведён полный пример: от развёртывания кластера Flink до чтения и записи данных Doris с помощью FlinkSQL. Прежде чем начать, выберите выпуск Connector, совместимый с вашими версиями Flink и Doris, согласно примечаниям к версиям, и получите пакет Jar, как описано в разделе Установка.

Рассмотрим на примере кластера Standalone:

  1. Скачайте установочный пакет Flink 1.18.1.
  2. После распаковки поместите пакет Jar Flink Doris Connector в <FLINK_HOME>/lib.
  3. Перейдите в каталог <FLINK_HOME> и выполните bin/start-cluster.sh, чтобы запустить кластер Flink.
  4. Используйте команду jps, чтобы убедиться, что кластер Flink успешно запущен.

2. Инициализация таблицы Doris​

Выполните следующий SQL-код, чтобы создать таблицы Doris и записать тестовые данные:

CREATE DATABASE test;

CREATE TABLE test.student (
    `id` INT,
    `name` VARCHAR(256),
    `age` INT
)
UNIQUE KEY(`id`)
DISTRIBUTED BY HASH(`id`) BUCKETS 1
PROPERTIES (
    "replication_allocation" = "tag.location.default: 3"
);

INSERT INTO test.student values(1, "James", 18);
INSERT INTO test.student values(2, "Emily", 28);

CREATE TABLE test.student_trans (
    `id` INT,
    `name` VARCHAR(256),
    `age` INT
)
UNIQUE KEY(`id`)
DISTRIBUTED BY HASH(`id`) BUCKETS 1
PROPERTIES (
    "replication_allocation" = "tag.location.default: 3"
);

3. Запуск задания FlinkSQL​

Запустите клиент FlinkSQL:

bin/sql-client.sh

Выполните следующий код FlinkSQL:

CREATE TABLE Student (
    id STRING,
    name STRING,
    age INT
)
WITH (
    'connector' = 'doris',
    'fenodes' = '127.0.0.1:8030',
    'table.identifier' = 'test.student',
    'username' = 'root',
    'password' = ''
);

CREATE TABLE StudentTrans (
    id STRING,
    name STRING,
    age INT
)
WITH (
    'connector' = 'doris',
    'fenodes' = '127.0.0.1:8030',
    'table.identifier' = 'test.student_trans',
    'username' = 'root',
    'password' = '',
    'sink.label-prefix' = 'doris_label'
);

INSERT INTO StudentTrans SELECT id, concat('prefix_', name), age + 1 FROM Student;

4. Запрос результата​

mysql> select * from test.student_trans;
+------+--------------+------+
| id   | name         | age  |
+------+--------------+------+
|    1 | prefix_James |   19 |
|    2 | prefix_Emily |   29 |
+------+--------------+------+
2 rows in set (0.02 sec)

Дальнейшие шаги​

Report issue