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

SeaTunnel

SeaTunnel — простая в использовании распределённая платформа интеграции данных со сверхвысокой производительностью. Она поддерживает синхронизацию больших объёмов данных в реальном времени и позволяет стабильно и эффективно синхронизировать десятки миллиардов записей в день.

Этот документ предназначен для пользователей, которым нужно синхронизировать данные из SeaTunnel с Doris. В нём для разных сценариев описано, как выбрать Doris Sink, настроить его параметры и использовать его.

Как выбрать коннектор​

Сценарий использованияРекомендуемый подходПоддерживаемая версия или движокОписание
Нужна запись с семантикой exactly-once или синхронизация данных CDCConnector-V2 Doris SinkSeaTunnel 2.3.1 и вышеПоддерживает Doris Sink, запись с семантикой exactly-once и синхронизацию данных CDC.
Синхронизация данных с Doris через движок FlinkConnector-V1 Flink Doris SinkSeaTunnel 2.1.0, движок FlinkПодходит для заданий SeaTunnel, которые уже выполняются на движке Flink.
Синхронизация данных с Doris через движок SparkConnector-V1 Spark Sink DorisSeaTunnel 2.1.0, движок SparkПодходит для заданий, которые уже выполняются на движке Spark, например для переноса данных из Hive в Doris.

Сценарий 1: запись в Doris с помощью Connector-V2​

Apache SeaTunnel Connector-V2 в SeaTunnel 2.3.1 поддерживает Doris Sink, запись с семантикой exactly-once и синхронизацию данных CDC.

Исходный код плагина доступен по ссылке код плагина SeaTunnel Doris Sink.

Настройка параметров​

ПараметрТипОбязательныйПо умолчаниюОписание
fenodesstringДа-Адреса узлов FE кластера Doris в формате fe_ip:fe_http_port,....
usernamestringДа-Имя пользователя Doris.
passwordstringДа-Пароль пользователя Doris.
table.identifierstringДа-Имя таблицы Doris в формате DBName.TableName.
sink.label-prefixstringДа-Префикс метки, используемый Stream Load. В сценарии с 2PC этот префикс должен быть глобально уникальным, чтобы гарантировать семантику EOS в SeaTunnel.
sink.enable-2pcboolНетtrueВключает двухфазную фиксацию (2PC). Значение по умолчанию — true, что гарантирует семантику exactly-once. Подробнее о двухфазной фиксации см. в руководстве по Stream Load.
sink.enable-deleteboolНетfalseВключает удаление. Для использования этого параметра в таблице Doris должно быть включено пакетное удаление. В Doris 0.15+ эта функция включена по умолчанию; поддерживается только модель таблиц Unique. Подробнее см. в разделе Пакетное удаление.
doris.configmapДа-Параметры data_desc для Stream Load. Дополнительные параметры приведены в руководстве по Stream Load.

Импорт данных в формате JSON​

sink {
    Doris {
        fenodes = "doris_fe:8030"
        username = root
        password = ""
        table.identifier = "test.table_sink"
        sink.enable-2pc = "true"
        sink.label-prefix = "test_json"
        doris.config = {
            format = "json"
            read_json_by_line = "true"
        }
    }
}

Импорт данных в формате CSV​

sink {
    Doris {
        fenodes = "doris_fe:8030"
        username = root
        password = ""
        table.identifier = "test.table_sink"
        sink.enable-2pc = "true"
        sink.label-prefix = "test_csv"
        doris.config = {
            format = "csv"
            column_separator = ","
            line_delimiter = "\n"
        }
    }
}

Сценарий 2: запись в Doris с помощью Connector-V1​

Apache SeaTunnel 2.1.0 поддерживает коннектор Doris, который позволяет синхронизировать данные с Doris через движок Flink или Spark.

Исходный код плагина доступен по ссылке код плагина SeaTunnel Flink Sink Doris.

Настройка параметров​

ПараметрТипОбязательныйПо умолчаниюПоддерживаемый движокОписание
fenodesstringДа-FlinkHTTP-адрес FE Doris, например 127.0.0.1:8030.
databasestringДа-FlinkИмя базы данных Doris, в которую выполняется запись.
tablestringДа-FlinkИмя таблицы Doris, в которую выполняется запись.
userstringДа-FlinkПользователь для доступа к Doris.
passwordstringДа-FlinkПароль пользователя для доступа к Doris.
batch_sizeintНет100FlinkМаксимальное число строк, записываемых в Doris одним пакетом.
intervalintНет1000FlinkИнтервал сброса буфера в миллисекундах. По истечении этого интервала асинхронный поток записывает данные из буфера в Doris. Установите значение 0, чтобы отключить периодическую запись.
max_retriesintНет1FlinkЧисло повторных попыток при ошибке записи в Doris.
doris.*stringНет-FlinkПараметры импорта Stream Load, например doris.column_separator = ','. Дополнительные параметры приведены в руководстве по Stream Load.

Пример: запись данных из сокета в Doris​

env {
    execution.parallelism = 1
}

source {
    SocketStream {
        host = 127.0.0.1
        port = 9999
        result_table_name = "socket"
        field_name = "info"
    }
}

transform {
}

sink {
    DorisSink {
        fenodes = "127.0.0.1:8030"
        user = root
        password = 123456
        database = test
        table = test_tbl
        batch_size = 5
        max_retries = 1
        interval = 5000
    }
}

Запустите задание:

sh bin/start-seatunnel-flink.sh --config config/flink.streaming.conf

Запись в Doris через движок Spark​

Исходный код плагина доступен по ссылке код плагина SeaTunnel Spark Sink Doris.

Настройка параметров​

ПараметрТипОбязательныйПо умолчаниюПоддерживаемый движокОписание
fenodesstringДа-SparkАдрес узла FE Doris, например 127.0.0.1:8030.
databasestringДа-SparkИмя базы данных Doris, в которую выполняется запись.
tablestringДа-SparkИмя таблицы Doris, в которую выполняется запись.
userstringДа-SparkПользователь для доступа к Doris.
passwordstringДа-SparkПароль пользователя для доступа к Doris.
batch_sizeintДа100SparkЧисло строк, отправляемых в каждом пакете при записи из Spark в Doris через Stream Load.
doris.*stringНет-SparkHTTP-параметры Stream Load. Чтобы использовать любой параметр Stream Load, добавьте к нему префикс doris., например doris.column_separator. Дополнительные параметры приведены в руководстве по Stream Load.

Пример: перенос данных из Hive в Doris​

env {
    spark.app.name = "hive2doris-template"
}

spark {
    spark.sql.catalogImplementation = "hive"
}

source {
    hive {
        preSql = "select * from tmp.test"
        result_table_name = "test"
    }
}

transform {
}

sink {
    Console {
    }

    Doris {
        fenodes = "xxxx:8030"
        database = "tmp"
        table = "test"
        user = "root"
        password = "root"
        batch_size = 1000
        doris.column_separator = "\t"
        doris.columns = "date_key,date_value,day_in_year,day_in_month"
    }
}

Запустите задание:

sh bin/start-waterdrop-spark.sh --master local[4] --deploy-mode client --config ./config/spark.conf
Report issue