Обзор

Seatunnel очень прост в использовании,высокая производительность、поддерживатьпрямая трансляцияиАвтономная пакетная обработкаПродукты для массовой обработки данных,Структурировано наApache Spark и Apache Flinkвыше,Адрес проекта с открытым исходным кодом: https://github.com/apache/incubator-seatunnel.

Эволюция версий

Seatunnel ранее был известен как Waterdrop.,После смены названия он был официально преобразован в проект Apache.,При этом два названия также соответствуют разным версиям.,Waterdrop относится к версии 1.x.,Seatunnel относится к версии 2.x.,Для 1.xи2.x имеются следующие отличия:

Зачем нам нужен Seatunnel

Apache Spark и Apache Flink — отличное достижение для распределенной и потоковой обработки данных.,Однако высокий порог использования требует от персонала, занимающегося обработкой данных, изучения сложного рабочего механизма и API Sparkiflink, чтобы использовать его более плавно.,Чтобы снизить порог обработки данных,И сделайте Sparkiflink проще в использовании,Сократите затраты на обучение,Ускорьте внедрение распределенной обработки данных в производственных средах,Появился Seatunnel.

Некоторые мысли, основанные на самых последних работах по обработке данных

1. Большая часть обработки данных повторяется
2. Код обработки данных избыточен
3. Доля работы по обработке данных, в которую входит синхронизация. данные работают, после завершения расчета автономного хранилища данных часто отключается синхронизация слоя рекламы. данные для предоставления функций внешнего отображения отчетов в базах данных OLAP (ck, es и т. д.), специально оптимизированных для запросов. Можно ли ускорить эти функции? Можно ли его использовать повторно?
4. В процессе обработки данных может возникнуть потребность в доступе к нескольким разнородным источникам данных, таким как файл, redis, hdfs, kafka, mysql... Как лучше справиться с требованиями интеграции разнородных источников данных?
5. На основе нынешнего появления все большего количества структур больших данных,Направление обработки больших данных постепенно меняется на SQL и low-code,С точки зрения бизнеса, независимо от того, сколько данных находится внизу, они будут заполнены в одной таблице или нескольких таблицах.,Если вы умеете использовать sql, вы можете рассчитывать огромные объемы данных и быстро получать правильные результаты.,Весь бизнес-отдел будет использовать данные более эффективно
6. Предположим, что предприятию необходимо настроить промежуточную платформу данных, как быстро предоставить возможности промежуточной платформы обработки данных внешнему миру?

Болевые точки бизнеса, которые может решить Seatunnel

1. Поддерживается двумя основными платформами распределенных данных: Spark и Flink.,Естественно, способен к распределенной обработке данных.,Предоставление предприятиям возможности больше сосредоточиться на извлечении ценности и обработке данных.,Вместо того, чтобы сосредоточиться на совместимости и развитии базовой технологии для больших данных
2. Используя совместимость распределенной инфраструктуры Spark и Flink для разнородных источников данных, можно добиться быстрой синхронизации и доступа к разнородным источникам данных.
3. Очень абстрактная логика бизнес-обработки,Уменьшите избыточность кода и дублируйте разработку.

Преимущества и недостатки подводного туннеля

Преимущества

1. Простота в использовании, гибкая настройка, не требует разработки.
2. Модульность и плагин
3. Поддерживает обработку и агрегирование данных с использованием SQL.
4. Благодаря своей высокоинкапсулированной архитектуре вычислительного механизма он может быть хорошо интегрирован со средней платформой и предоставлять возможности распределенных вычислений внешнему миру.

недостаток

1. Искра поддерживает версию 2.2.0 - 2.4.8,Нетподдержка искры3.x
2. Flink поддерживает версию 1.9.0. В настоящее время Flink обновлен до версии 1.14.x и не имеет обратной совместимости.
3. Хотя задания Spark можно настроить быстро, соответствующий персонал все равно должен знать, как настроить некоторые параметры, чтобы сделать задание более эффективным.

Сопутствующие конкурирующие продукты и сравнения

• FlinkX, теперь переименованный в Чунджун
• StreamX
• DataX

Основные концепции и основные принципы Seatunnel

основные понятия

1. Суть всей конструкции Seatunnel заключается в использовании в шаблоне проектирования «переворота управления» или «внедрения зависимостей».,В основном резюмируется в следующих двух пунктах：
   1. Верхний уровень не зависит от нижнего уровня, оба полагаются на абстракцию.
   2. Код процесса и бизнес-логика должны быть разделены.
2. Для всего процесса обработки данных,Условно его можно разделить на следующие процессы:：входить -> Конвертировать -> выход,Для более сложной обработки данных,По сути, это также комбинация следующих типов поведения:

1. Seatunnel обеспечивает высокую степень абстракции для этих типов обработки данных. На основе этой относительно полной абстракции 80% повторяющихся операций в процессе обработки бизнес-данных выделяются и преобразуются в подключаемые модули с возможностью горячей замены, что позволяет бизнесу выполнять горячую замену. Логика обработки отделена от всей обработки данных, что позволяет пользователям больше сосредоточиться на реализации и внедрении бизнеса.

Основной принцип

1. Динамическая регистрация плагина использует технологию Java SPI для обеспечения гибкого расширения структуры. Идеи дизайна относятся к Presto, ES и т. д. Заинтересованные студенты могут пойти и учиться самостоятельно. es использует Google Guice, а Presto использует. Java, упомянутый выше.
2. На основе вышеизложенной теории,Обработка данных требует единой абстракции и преобразования.,Что очень хорошо подходит, так это то, что искра или мигание уже сделали эту работу за нас.,спаркдатасет,DataSet и DataStream компании Flink уже представляют собой высокую степень абстракции для доступа к данным.,По сути, обработка данных — это обработка этих структур данных.,В то же время эти данные могут быть зарегистрированы в виде таблиц в контексте после доступа к ним.,SQL можно использовать для обработки на основе таблиц.
3. Весь Seatunnel генерирует задание искры или задание флинка через файл конфигурации.
4. Стек технологий включает в себя следующие：
   1. Java
   2. Scala
   3. Flink
   4. Spark
   5. Java spi

Проектирование архитектуры подключаемого модуля Spark

Проектирование архитектуры плагина Flink

Ход выполнения программы

Подробности реализации абстракции подключаемого модуля верхнего уровня

public interface Plugin<T> extends Serializable {
    // Ключ файла конфигурации
    String RESULT_TABLE_NAME = "result_table_name";
    String SOURCE_TABLE_NAME = "source_table_name";
    
    // Установите конфигурацию для каждого плагина
    void setConfig(Config config);
    
    // Получить конфигурацию плагина 
    Config getConfig();
    
    // Проверка конфигурации
    CheckResult checkConfig();
    
    // Подготовка перед подключением
    void prepare(T prepareEnv);
}

Подробности реализации абстракции верхнего уровня подключаемого модуля Spark

Пакетная обработка Пакетная обработка

Source

trait BaseSparkSource[Data] extends BaseSource[SparkEnvironment] {

  protected var config: Config = ConfigFactory.empty()

  override def setConfig(config: Config): Unit = this.config = config

  override def getConfig: Config = config

  def getData(env: SparkEnvironment): Data;

}

Transform

trait BaseSparkTransform extends BaseTransform[SparkEnvironment] {

  protected var config: Config = ConfigFactory.empty()

  override def setConfig(config: Config): Unit = this.config = config

  override def getConfig: Config = config

  def process(data: Dataset[Row], env: SparkEnvironment): Dataset[Row];

}

Output

trait BaseSparkSink[OUT] extends BaseSink[SparkEnvironment] {

  protected var config: Config = ConfigFactory.empty()

  override def setConfig(config: Config): Unit = this.config = config

  override def getConfig: Config = config

  def output(data: Dataset[Row], env: SparkEnvironment): OUT;

}

Потоковая обработкаStream

trait SparkStreamingSource[T] extends BaseSparkSource[DStream[T]] {

  def beforeOutput(): Unit = {}

  def afterOutput(): Unit = {}

  def rdd2dataset(sparkSession: SparkSession, rdd: RDD[T]): Dataset[Row]

  def start(env: SparkEnvironment, handler: Dataset[Row] => Unit): Unit = {
    getData(env).foreachRDD(rdd => {
      val dataset = rdd2dataset(env.getSparkSession, rdd)
      handler(dataset)
    })
  }

}

Подробности реализации абстракции верхнего уровня подключаемого модуля Flink

Пакетная обработка Пакетная обработка

Source

public interface FlinkBatchSource<T> extends BaseFlinkSource {

    DataSet<T> getData(FlinkEnvironment env);
}

Transform

public interface FlinkBatchTransform<IN, OUT> extends BaseFlinkTransform {

    DataSet<OUT> processBatch(FlinkEnvironment env, DataSet<IN> data);

}

Output

public interface FlinkBatchSink<IN, OUT> extends BaseFlinkSink {

    DataSink<OUT> outputBatch(FlinkEnvironment env, DataSet<IN> inDataSet);

}

Потоковая обработкаStream

Source

public interface FlinkStreamSource<T> extends BaseFlinkSource {

    DataStream<T> getData(FlinkEnvironment env);

}

Transform

public interface FlinkStreamTransform<IN, OUT> extends BaseFlinkTransform {

    DataStream<OUT> processStream(FlinkEnvironment env, DataStream<IN> dataStream);
}

Output

public interface FlinkStreamSink<IN, OUT> extends BaseFlinkSink {

    DataStreamSink<OUT> outputStream(FlinkEnvironment env, DataStream<IN> dataStream);

}

Пользовательские шаги плагина

1. Наследовать соответствующие интерфейсы для разных платформ и типов плагинов.,Основные методы обработки в интерфейсе
2. Зарегистрируйтесь в Java Spi
3. Поместите свой собственный определенный jar-пакет в каталог плагинов основного jar-пакета Seatunnel.

Анализ принципов Java spi

концепция

Полное имя SPI Сервис Provider Interface,Это набор интерфейсов, предоставляемых Java, которые реализованы или расширены третьими сторонами.,Его можно использовать для включения расширений платформы и замены компонентов.,Роль SPI — найти реализации сервисов для этих расширенных API.

Разница между API и SPI

API-(Application Programming Interface)большую часть времени,ВсеразработчикРазработать интерфейс и завершить его реализацию,звонящийПолагайтесь только на вызовы интерфейса,и не имеет права выбирать другую реализацию. С точки зрения пользователей,API используется непосредственно разработчиками приложений.,SPI-(Service Provider Interface)дазвонящийразработать спецификации интерфейса,Предоставляется сторонним сторонам для реализациизвонящийВыберите нужную вам внешнюю реализацию。С точки зрения пользователей,SPI Используется расширителями фреймворка

Реализация демо

1. Определить интерфейс

 package com.tyrantlucifer;
 
 public interface Animal {
     void shut();
 }

1. Определите основную функцию и используйте загрузчик служб для динамической загрузки.

 package com.tyrantlucifer;
 
 import java.util.ServiceLoader;
 
 public class Main {
     public static void main(String[] args) {
         ServiceLoader<Animal> services = ServiceLoader.load(Animal.class);
         for (Animal service : services) {
             service.shut();
         }
     }
 }

1. реализовать интерфейс

 package com.tyrantlucifer;
 
 public class Cat implements Animal {
     public void shut() {
         System.out.println("cat shut miao miao!!!");
     }
 }

 package com.tyrantlucifer;
 
 public class Dog implements Animal{
     public void shut() {
         System.out.println("dog shut wang wang!!!");
     }
 }

Зарегистрировать Спи,Вам необходимо создать новый файл с полным именем класса интерфейса в разделе resources/META-INF/services.,Например, наш интерфейс на этот разcom.tyrantlucifer.Animal,Затем создайте новыйcom.tyrantlucifer.Animalдокумент,И добавьте в файл свой собственный класс реализации:

com.tyrantlucifer.Cat
com.tyrantlucifer.Dog

Демо-версия SeatunnelДемо

1. Spark

 spark {
   spark.streaming.batchDuration = 5
   spark.app.name = "seatunnel"
   spark.ui.port = 13000
 }
 
 input {
   socketStream {}
 }
 
 filter {
   split {
     fields = ["msg", "name"]
     delimiter = ","
   }
 }
 
 output {
   stdout {}
 }

1. Flink

 env {
   execution.parallelism = 1
 }
 
 source {
     SocketStream{
           result_table_name = "fake"
           field_name = "info"
     }
 }
 
 transform {
   Split{
     separator = "#"
     fields = ["name","age"]
   }
   sql {
     sql = "select * from (select info,split(info) as info_row from fake) t1"
   }
 }
 
 sink {
   ConsoleSink {}
 }

1. Пользовательский плагин

 class MyStdout extends BaseOutput {
 
   var config: Config = ConfigFactory.empty()
 
   /**
    * Set Config.
    * */
   override def setConfig(config: Config): Unit = {
     this.config = config
   }
 
   /**
    * Get Config.
    * */
   override def getConfig(): Config = {
     this.config
   }
 
   override def checkConfig(): (Boolean, String) = {
     if (!config.hasPath("limit") || (config.hasPath("limit") && config.getInt("limit") >= -1)) {
       (true, "")
     } else {
       (false, "please specify [limit] as Number[-1, " + Int.MaxValue + "]")
     }
   }
 
   override def prepare(spark: SparkSession): Unit = {
     super.prepare(spark)
 
     val defaultConfig = ConfigFactory.parseMap(
       Map(
         "limit" -> 100,
         "format" -> "plain" // plain | json | schema
       )
     )
     config = config.withFallback(defaultConfig)
   }
 
   override def process(df: Dataset[Row]): Unit = {
 
     val limit = config.getInt("limit")
 
     var format = config.getString("format")
     if (config.hasPath("serializer")) {
       format = config.getString("serializer")
     }
     format match {
       case "plain" => {
         if (limit == -1) {
           df.show(Int.MaxValue, false)
         } else if (limit > 0) {
           df.show(limit, false)
         }
       }
       case "json" => {
         if (limit == -1) {
           df.toJSON.take(Int.MaxValue).foreach(s => println(s))
 
         } else if (limit > 0) {
           df.toJSON.take(limit).foreach(s => println(s))
         }
       }
       case "schema" => {
         df.printSchema()
       }
     }
   }
 }

Q&A