apache/seatunnel

SeaTunnel is a next-generation super high-performance, distributed, massive data integration tool.

Apache SeaTunnel

Table of Contents

• Overview
• Why Choose SeaTunnel
• Key Features
• SeaTunnel Workflow
• Supported Connectors
• Getting Started
• Use Cases
• Code of Conduct
• Contributors
• How to Compile
• Contact Us
• Landscapes
• Apache SeaTunnel Web Project
• Our Users
• License
• Frequently Asked Questions

Overview

SeaTunnel is a next-generation, high-performance, distributed data integration tool, capable of synchronizing vast amounts of data daily. It's trusted by numerous companies for its efficiency and stability.

Why Choose SeaTunnel

SeaTunnel addresses common data integration challenges:

• Diverse Data Sources: Seamlessly integrates with hundreds of evolving data sources.

• Complex Synchronization Scenarios: Supports various synchronization methods, including real-time, CDC, and full database synchronization.

• Resource Efficiency: Minimizes computing resources and JDBC connections for real-time synchronization.

• Quality and Monitoring: Provides data quality and monitoring to prevent data loss or duplication.

Key Features

• Diverse Connectors: Offers support for over 100 connectors, with ongoing expansion.

• Batch-Stream Integration: Easily adaptable connectors simplify data integration management.

• Distributed Snapshot Algorithm: Ensures data consistency across synchronized data.

• Multi-Engine Support: Works with SeaTunnel Zeta Engine, Flink, and Spark.

• JDBC Multiplexing and Log Parsing: Efficiently synchronizes multi-tables and databases.

• High Throughput and Low Latency: Provides high-throughput data synchronization with low latency.

• Real-Time Monitoring: Offers detailed insights during synchronization.

• Two Job Development Methods: Supports coding and visual job management with the SeaTunnel web project.

SeaTunnel Workflow

Configure jobs, select execution engines, and parallelize data using Source Connectors. Easily develop and extend connectors to meet your needs.

Supported Connectors

• Source Connectors
• Sink Connectors
• Transform Connectors

For a list of connectors and their health status, visit the Connector Status.

Getting Started

Download SeaTunnel from the official website.

Choose your runtime execution engine:

• SeaTunnel Zeta Engine
• Spark
• Flink

Use Cases

Explore real-world use cases of SeaTunnel, such as Weibo, Tencent Cloud, Sina, Sogou, and Yonghui Superstores. More use cases can be found on the SeaTunnel blog.

Code of Conduct

Participate in this project following the Contributor Covenant Code of Conduct.

Contributors

We appreciate all developers for their contributions. See the list of contributors.

How to Compile

Refer to this document for compilation instructions.

Contact Us

• Mail list: [email protected]. Subscribe by sending an email to [email protected].

• Slack: Join SeaTunnel Slack

• Twitter: ASFSeaTunnel on Twitter

Landscapes

SeaTunnel enriches the CNCF CLOUD NATIVE Landscape.

Apache SeaTunnel Web Project

SeaTunnel Web is a web project that provides visual management of jobs, scheduling, running and monitoring capabilities. It is developed based on the SeaTunnel Connector API and the SeaTunnel Zeta Engine. It is a web project that can be deployed independently. It is also a sub-project of SeaTunnel. For more information, please refer to SeaTunnel Web

Our Users

Companies and organizations worldwide use SeaTunnel for research, production, and commercial products. Visit our user page for more information.

License

Apache 2.0 License

Frequently Asked Questions

1. How do I install SeaTunnel?

Follow the installation guide on our website to get started.

2. How can I contribute to SeaTunnel?

We welcome contributions! Please refer to our Contribution Guidelines for details.

3. How do I report issues or request features?

You can report issues or request features on our GitHub repository.

4. Can I use SeaTunnel for commercial purposes?

Yes, SeaTunnel is available under the Apache 2.0 License, allowing commercial use.

5. Where can I find documentation and tutorials?

Our official documentation includes detailed guides and tutorials to help you get started.

7. Is there a community or support channel?

Join our Slack community for support and discussions: SeaTunnel Slack.

seatunnel's Issues

Antlr4 tutorials

Antlr tutorial:

https://tomassetti.me/antlr-mega-tutorial/
http://sqtds.github.io/tags/antlr4/
https://alexecollins.com/antlr4-and-maven-tutorial/
http://meri-stuff.blogspot.com/2011/09/antlr-tutorial-expression-language.html#LexerBasics
http://progur.com/2016/09/how-to-create-language-using-antlr4.html
https://yq.aliyun.com/articles/11366
http://www.cnblogs.com/sld666666/p/6145854.html
http://blog.csdn.net/dc_726/article/details/45399371
https://github.com/antlr/antlr4/blob/master/doc/index.md
https://github.com/antlr/grammars-v4/blob/master/json/JSON.g4
https://plugins.jetbrains.com/plugin/7358-antlr-v4-grammar-plugin
https://stackoverflow.com/questions/21534316/is-there-a-simple-example-of-using-antlr4-to-create-an-ast-from-java-source-code
https://stackoverflow.com/questions/23092081/antlr4-visitor-pattern-on-simple-arithmetic-example

https://stackoverflow.com/questions/6487593/what-does-fragment-mean-in-antlr
http://floris.briolas.nl/floris/2008/10/antlr-common-pittfals/
https://github.com/odiszapc/nginx-java-parser
https://codevomit.wordpress.com/2015/04/25/antlr4-project-with-maven-tutorial-episode-3/
https://stackoverflow.com/questions/1931307/antlr-is-there-a-simple-example
https://stackoverflow.com/questions/29971097/how-to-create-ast-with-antlr4

Listener vs Vistor:

https://stackoverflow.com/questions/20714492/antlr4-listeners-and-visitors-which-to-implement?rq=1

http://jakubdziworski.github.io/java/2016/04/01/antlr_visitor_vs_listener.html

ANTLRv4: How to read double quote escaped double quotes in string?

https://stackoverflow.com/questions/17897651/antlrv4-how-to-read-double-quote-escaped-double-quotes-in-string

nested boolean expression parsing:

https://stackoverflow.com/questions/25096713/parser-lexer-logical-expression
https://stackoverflow.com/questions/30976962/nested-boolean-expression-parser-using-antlr

parsing comment:

https://stackoverflow.com/questions/7070763/parse-comment-line?rq=1
https://stackoverflow.com/questions/28674875/antlr-4-how-to-parse-comments
http://meri-stuff.blogspot.com/2012/09/tackling-comments-in-antlr-compiler.html

design pattern: visitor

https://dzone.com/articles/design-patterns-visitor

books:

"The Definitive Antlr4 Reference"

scala code generation and runtime compile

http://yefremov.net/blog/scala-code-generation/

http://eed3si9n.com/treehugger/

https://stackoverflow.com/questions/23874281/scala-how-to-compile-code-from-an-external-file-at-runtime

https://gist.github.com/metasim/7492509

https://stackoverflow.com/questions/39137175/dynamically-compiling-scala-class-files-at-runtime-in-scala-2-11

https://stackoverflow.com/questions/12122939/generating-a-class-from-string-and-instantiating-it-in-scala-2-10/12123609#12123609

https://scalerablog.wordpress.com/2016/06/20/scala-code-interpretation-at-runtime/

https://scala-lms.github.io/

https://scala-lms.github.io/

https://github.com/codacy/codacy-macros

scala good coding style

• Codacy Coding Style: Enforce usage of the Try object[disabled]

http://www.scala-lang.org/api/current/scala/util/Try.html

https://stackoverflow.com/questions/25588387/scala-util-try-how-to-get-throwable-value-pattern-matching

中文／英文文档

中文文档完成度：

• 欢迎页 https://interestinglab.github.io/waterdrop
• 介绍 https://interestinglab.github.io/waterdrop/#/zh-cn/README
• (Garyelephant, 1月3日) 快速开始 https://interestinglab.github.io/waterdrop/#/zh-cn/quick-start
• (Garyelephant, 1月3日) 安装 https://interestinglab.github.io/waterdrop/#/zh-cn/installation

---

• 配置

• (Garyelephant) 通用配置

• Input插件

  • Fake
  • File
  • Hdfs
  • Kafka
  • S3
  • Socket

• Filter插件

  • (Rickyhuo)Add
  • Checksum
  • (Rickyhuo)Convert
  • (Rickyhuo)Date
  • Drop
  • Geoip
  • Grok
  • Json
  • Kv
  • (Rickyhuo)Lowercase
  • (Rikcyhuo)Remove
  • (Rickyhuo)Rename
  • Repartition
  • Replace
  • Sample
  • Split
  • SQL
  • Table
  • Truncate
  • (Rickyhuo)Uppercase
  • Uuid

• Output插件

  • Elasticsearch
  • File
  • Hdfs
  • Jdbc
  • (Rickyhuo)Kafka
  • MySQL
  • S3
  • Stdout

---

• 部署(Garyelephant, 1月10日)
• 监控(Garyelephant)
• 性能与调优
• (Rickyhuo, 1月2日) 插件开发
• (Garyelephant, 1月10日) Roadmap
• 贡献代码

---

英文文档完成度：

自动Build体系

travis.ci

20170904Week TODO

我列举一下本周可以做的事情：

(1) Filter UDF：

a) 找到Spark SQL自带的所有UDF列表，看看这些UDF都能做什么，将来我们可以在Waterdrop的文档里引用这些用户可以直接用的UDF；

b) 我们计划实现的那些Filter，能不能同时提供对应的UDF，如果能该怎么做？

c) 我们的Filter与SparkSQL自带的UDF有没有重复，能不能复用。

(2) 确定 BaseFilter的最终接口定义；

一个思路是把所有的Filter插件整理一遍，看它们需要什么样的BaseFilter接口定义；

另一个思路是想明白，如果将来有一个用户要开发他自己的插件，他该如何利用BaseFilter的接口开发出自己的插件。

(3) 在流程代码中支持多个 input, output

(4) [先搞定前3个再看此条] 确定BaseInput, BaseOutput的接口定义，这个涉及到几个麻烦的技术点，下周再跟你说。

Spark Streaming & Spark SQL

RDD vs DataFrame vs Dataset:

https://stackoverflow.com/questions/37301226/difference-between-dataset-api-and-dataframe
https://dzone.com/articles/the-dominant-apis-of-spark-datasets-dataframes-and

http://spark.apache.org/docs/latest/streaming-programming-guide.html
https://databricks.com/blog/2016/07/14/a-tale-of-three-apache-spark-apis-rdds-dataframes-and-datasets.html
http://www.csdn.net/article/2015-04-03/2824407
https://stackoverflow.com/questions/29383578/how-to-convert-rdd-object-to-dataframe-in-spark
https://indatalabs.com/blog/data-engineering/convert-spark-rdd-to-dataframe-dataset
https://databricks.com/blog/2016/01/04/introducing-apache-spark-datasets.html

项目框架(Event, 数据流, Dataframe,UDF, UDAF)的开发与测试

当前遇到的困难

• Derive multiple columns from a single column in a Spark DataFrame／Assign the result of UDF to multiple dataframe columns:

https://stackoverflow.com/questions/32196207/derive-multiple-columns-from-a-single-column-in-a-spark-dataframe

https://stackoverflow.com/questions/35322764/apache-spark-assign-the-result-of-udf-to-multiple-dataframe-columns

https://blog.cloudera.com/blog/2017/02/working-with-udfs-in-apache-spark/

https://stackoverflow.com/questions/32196207/derive-multiple-columns-from-a-single-column-in-a-spark-dataframe

configParse丢掉配置中最后一个字符

filter.Sql插件中sql字段读缺失最后一个字符

app.conf

spark {
  spark.streaming.batchDuration = 5
  spark.master = "local[2]"
  spark.app.name = "Waterdrop-1"
  spark.ui.port = 13000
}

input {
  kafka {
        topics = "sinabip_test"
        consumer.auto.offset.reset = "largest"
    }
}

filter {
	Split {
		source_field = "raw_message"
		fields = ["times", "info"]
	}
	Sql {
		table_name = "test",
		sql = "select info from test where info='hello'"
	}
	
}

output {
  Stdout {}
}

Main函数中打印的配置

[INFO] Parsed Config: 
{
    "filter" : [
        {
            "entries" : {
                "fields" : [
                    "times",
                    "info"
                ],
                "source_field" : "raw_message"
            },
            "name" : "Split"
        },
        {
            "entries" : {
                "sql" : "select info from test where info='hello",
                "table_name" : "test"
            },
            "name" : "Sql"
        }
    ],
    "input" : [
        {           
            "name" : "kafka"
        }
    ],
    "output" : [
        {
            "entries" : {},
            "name" : "Stdout"
        }
    ],
    "spark" : {
        "spark" : {
            "app" : {
                "name" : "Waterdrop-1"
            },
            "master" : "local[2]",
            "streaming" : {
                "batchDuration" : 5
            },
            "ui" : {
                "port" : 13000
            }
        }
    }
}

conditional and expression in configurtion file

example:

if [a.b.c != 3] {
	
} else {
	
}

references:

logstash configuration
http://enear.github.io/2016/03/31/parser-combinators/
https://github.com/sirthias/parboiled2
http://www.lihaoyi.com/fastparse/
https://gist.github.com/nicerobot/4189552
https://dzone.com/articles/rolling-your-own-dsl-in-scala
https://www.slideshare.net/abhijit.sharma/writing-dsls-in-scala
https://www.slideshare.net/holograph/a-field-guide-to-dsl-design-in-scala
http://jmespath.org/
http://www.antlr.org/

Search keyword: Grammars, workflows, DSL, expression, eval

文档的书写格式确定及文档生成工具的确定

markdown + docsify

自动化测试体系

Scala Unit Test:

http://www.scalatest.org/
https://alvinalexander.com/scala/writing-tdd-unit-tests-with-scalatest
https://semaphoreci.com/community/tutorials/a-hands-on-introduction-to-scalatest

新增src/test/java目录，其下main方法运行报错

想新建一个测试目录 src/test/java，新建一个类的Main方法

org.interestinglab.waterdrop.WaterDropTest

运行报错

Error:(13, 14) BoolExprBaseVisitor is already defined as object BoolExprBaseVisitor
public class BoolExprBaseVisitor<T> extends AbstractParseTreeVisitor<T> implements BoolExprVisitor<T> {
             ^

Scala Unit Test

http://www.scalatest.org/
https://alvinalexander.com/scala/writing-tdd-unit-tests-with-scalatest
https://semaphoreci.com/community/tutorials/a-hands-on-introduction-to-scalatest

Spark RDD处理多语言支持

You can use the pipe() function on RDDs to call external code. It passes data to an external program through stdin / stdout. For Spark Streaming, you would do dstream.transform(rdd => rdd.pipe(...)) to call it on each RDD.

支持用户指定字段类型StructType

插件体系构建(包括UDF, UDAF体系)

插件体系和具体的实现可以参考

Cloudera Morphlines
Embulk
Fluent
Hangout
Logstash
Kinesis
Spark UDF
Hive UDF

---

Filter插件需求：
Distinct，sample

---

用户需求：
repartition 增加或减少并行度，输出文件个数。

monitoring

• JMX

• Coda Hale Metrics

https://developer.lightbend.com/docs/monitoring/2.3.x/plugins/chmetrics/intro.html

https://www.slideshare.net/IzzetMustafaiev/metrics-by-coda-hale

http://metrics.dropwizard.io/3.2.3/getting-started.html

易用性提升的几点备注

1. debug 模式：能让用户很容易知道每个环节的数据变化

2. 本地模式：利用spark 的local模式，方便用户debug和本地开发。

3. Filter过程可视化。

4. 帮用户想好应用场景，并简化对应的部署和运行流程。

同时支持国内外IP库

各种IP库，我们除了要支持geoip2,还需要挑选1个国内的ip库做支持：
ipip.net【国外不准，国内准】
纯真ip数据库【国外不准，国内准】 http://www.cz88.net/， http://www.cnblogs.com/anpengapple/p/5384985.html
geoip2【国外准，国内不准】

Spark Logging Configuration

https://community.hortonworks.com/articles/138849/how-to-capture-spark-driver-and-executor-logs-in-y.html

https://hackernoon.com/how-to-log-in-apache-spark-f4204fad78a

https://mapr.com/blog/how-log-apache-spark/

配置中无法指定\s+等类似的正则表达式

主要涉及到的插件包括filter: split, table

Document

https://interestinglab.github.io/waterdrop/#/zh-cn/README

Description

Filter插件开发

插件体系和具体的实现可以参考

插件体系和具体的实现可以参考

logstash
hangout
Spark SQL UDF,UDAF
Hive UDF
flume
fluent
kafka stream

https://github.com/onurakpolat/awesome-bigdata#data-ingestion

---

Filter插件需求：
Distinct，sample

---

用户需求：
repartition 增加或减少并行度，输出文件个数。

coding style 优化

• 加入java的checkstyle检查，
• scalastyle配置更严格的coding style
• codacy做对应的配置

配置中支持复杂配置逻辑（如：if else 的逻辑，模版变量，预定义变量）

conditional and expression in configurtion file:

example:

if [a.b.c != 3] {
	
} else {
	
}

references:

logstash configuration
http://enear.github.io/2016/03/31/parser-combinators/
https://github.com/sirthias/parboiled2
http://www.lihaoyi.com/fastparse/
https://gist.github.com/nicerobot/4189552
https://dzone.com/articles/rolling-your-own-dsl-in-scala
https://www.slideshare.net/abhijit.sharma/writing-dsls-in-scala
https://www.slideshare.net/holograph/a-field-guide-to-dsl-design-in-scala
http://jmespath.org/
http://www.antlr.org/

Search keyword: Grammars, workflows, DSL, expression, eval

Structured Streaming & Window Operations

https://databricks.com/blog/2016/07/28/structured-streaming-in-apache-spark.html

https://spark-summit.org/2016/events/a-deep-dive-into-structured-streaming/

https://databricks.com/blog/2017/05/18/taking-apache-sparks-structured-structured-streaming-to-production.html

https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html

https://www.slideshare.net/julesdamji/a-deep-dive-into-structured-streaming-in-apache-spark

http://www.jianshu.com/p/2673a6e4254f

filterConfObj.foreach遍历顺序不可控

filterConfObj.foreach遍历顺序不是按照配置文件中插件顺序从上至下遍历

configParse对正则表达式支持不完善

1. 使用\s

   pattern = "\s"

   运行报错

2. 使用\\s

   pattern = "\s"

   configParse会对\转义

向后兼容性

• spark升级到2.2之后由于scala升级到2.11导致无法兼容spark1.6

• 兼容jdk 1.7

Structured Streaming

https://databricks.com/blog/2016/07/28/continuous-applications-evolving-streaming-in-apache-spark-2-0.html

Spark Datasource API Integration

https://docs.databricks.com/spark/latest/data-sources/index.html

https://www.slideshare.net/databricks/yin-huai-20150325meetupwithdemos

https://mapr.com/blog/spark-data-source-api-extending-our-spark-sql-query-engine/

https://jaceklaskowski.gitbooks.io/mastering-apache-spark/spark-sql-datasource-api.html

http://www.spark.tc/exploring-the-apache-spark-datasource-api/

https://mapr.com/blog/how-integrate-custom-data-sources-apache-spark/

Document 增加内部原理的介绍

Output插件开发

Spark UI(统计功能，各阶段耗时，Pipeline Viewer)

Document

https://interestinglab.github.io/waterdrop/#/

Description

网站文档体验优化

增加页面评论／讨论功能，如disqus

增加实时反馈：
https://www.hotjar.com

未开发插件

• Input

Flume, Http

• Filter

Clone, Dict, Geoip

• Output

Http

Roadmap

Waterdrop 未来5个重要的发展方向：

• 支持Flink/Bean 计算引擎

• 支持基于Flink 的【有状态】【实时】【聚合】计算(用户可指定时间粒度，纬度，指标)

• 交互式 UI（支持Pipeline的交互式构建、交互式的SQL执行，功能和性能诊断可视化工具）

• 基于应用场景的深入Spark/Flink 底层做性能优化。

• 扩大生产环境使用Waterdrop的公司规模（国内公司技术支持，英文社区推广）

• 自助化和交互式的问题诊断和性能优化，参考Alibaba Arthas

Document

https://interestinglab.github.io/waterdrop/#/zh-cn/configuration/output-plugins/Stdout

Description

从一个udf中生成多个新的column

应用场景grok, split等filter

支持在start-waterdrop.sh命令参数中使用spark配置

支持在start-waterdrop.sh命令参数中使用spark配置，如运行资源的大小，优先级高于application.conf ，类似spark-submit。

Input主要插件开发

scala code format

scala code format:
http://scalameta.org/scalafmt/

document generator

document:
https://github.com/QingWei-Li/docsify/#showcase

JSON插件目前的问题

1. 当target_field为***ROOT***时， 由于需要从原始数据获取schema，导致目前没有解决办法
2. 嵌套JSON结构无法多级解析

插件体系的结构问题

当前目录结构

filter/
├── BaseFilter.scala
└── Split.scala
└── Sql.scala

在具体文件上面加一级document会不会好点， 因为有的插件代码不可能都在一个文件里面完成，如果都放在一起的话， 不好管理。以这个grok为例

filter/
├── BaseFilter.scala
├── grok
│   ├── Grok.scala
│   └── PatternGrok.scala
├── split
│   └── Split.scala
└── sql
    └── Sql.scala

Document

https://interestinglab.github.io/waterdrop/#/zh-cn/configuration/filter-plugins/SQL

Description

M1 TODO

• 配置解析

  • spark common config(spark.*，还有appname, duration)解析
  • 配置文件错误提示和定位
  • 【非必需】实现if..else逻辑的代码【与插件流程体系直接相关】
  • 【非必需】用户预定义模版变量，系统环境变量替换

• 插件流程体系

  • 确定 BaseFilter最终接口定义（重点：filter(包括其他开发者的filter)根据需要自动注册为UDF）
  • 确定BaseInput, BaseOutput的接口定义(考虑到broadcast, accumulator 的应用；与spark input,output format的关系)
  • 在流程代码中支持多个 input, output
  • Serializer 与其他Plugin的关系
  • 能够集成外部开发者的插件（支持：Java/Scala）
  • 【非必需】Field Reference
  • 【非必需】支持if..else逻辑

• Input，Filter，Output插件开发

  • Input 插件
  • Filter 插件
  • Output插件
  • Input, Filter, Output插件功能测试(spark on Yarn[client, cluster]模式，spark on Mesos, Local)

• 全流程简化

  • 区分不同的build.sbt
  • 接管整个spark + waterdrop 的流程。同时允许waterdrop以最简单spark job方式运行。
  • 安装
  • 部署(3种部署方式)
  • 插件集成
  • 配置
  • 运行

• 中英文文档

  • 统一的插件定义的文档格式
  • 完整的中文文档（重点插件文档）
  • 完整的英文文档（重点插件文档）

[在这个节点上线]

---

• 性能报告
  • 【非必需】大数据量的稳定性，处理性能，一致性的测试。
  • 【非必需】性能报告
  • 【非必需】性能调优（并行度，filter体系代码）