Apache SeaTunnel

https://seatunnel.incubator.apache.org/zh-CN/docs/2.3.13/
Admin
Apache SeaTunnel is a distributed data integration platform that supports real-time high-volume data...
Tokens:312,589
Snippets:2,990
Trust Score:10
Update:2 months ago
Show doc for...
Tokens:
Raw
### Configure and Install Connectors

Source: https://seatunnel.incubator.apache.org/zh-CN/docs/2.3.13/getting-started/locally/quick-start-seatunnel-engine

Define the required connectors in the configuration file and execute the installation script.

```text
# 配置连接器名称  
--seatunnel-connectors--  
connector-jdbc  
connector-doris  
--end--  
```

```bash
# 安装连接器  
sh bin/install-plugin.sh  
```

--------------------------------

### MatchQuery Syntax Examples

Source: https://seatunnel.incubator.apache.org/zh-CN/docs/2.3.13/connectors/source/MongoDB

Examples of query syntax for various data types.

```text
# Query Boolean type  
"{c_boolean:true}"  
# Query string type  
"{c_string:\"OCzCj\"}"  
# Query the integer  
"{c_int:2}"  
# Type of query time  
"{c_date:ISODate(\"2023-06-26T16:00:00.000Z\")}"  
# Query floating point type  
{c_double:{$gte:1.71763202185342e+308}}  
```

--------------------------------

### Basic Redis Configuration Examples

Source: https://seatunnel.incubator.apache.org/zh-CN/docs/2.3.13/connectors/source/Redis

Examples showing simple Redis source configurations with text and JSON formats.

```hocon
Redis {  
  host = localhost  
  port = 6379  
  keys = "key_test*"  
  data_type = key  
  format = text  
}  
```

```hocon
Redis {  
  host = localhost  
  port = 6379  
  keys = "key_test*"  
  data_type = key  
  format = json  
  schema {  
    fields {  
      name = string  
      age = int  
    }  
  }  
}  
```

--------------------------------

### Easysearch SSL Configuration Examples

Source: https://seatunnel.incubator.apache.org/zh-CN/docs/2.3.13/connectors/source/Easysearch

Examples for configuring SSL/TLS connections, including disabling certificate or hostname verification and using keystores.

```hocon
source {  
    Easysearch {  
        hosts = ["https://localhost:9200"]  
        username = "admin"  
        password = "admin"  
          
        tls_verify_certificate = false  
    }  
}  
```

```hocon
source {  
    Easysearch {  
        hosts = ["https://localhost:9200"]  
        username = "admin"  
        password = "admin"  
          
        tls_verify_hostname = false  
    }  
}  
```

```hocon
source {  
    Easysearch {  
        hosts = ["https://localhost:9200"]  
        username = "admin"  
        password = "admin"  
          
        tls_keystore_path = "${your Easysearch home}/config/certs/http.p12"  
        tls_keystore_password = "${your password}"  
    }  
}  
```

--------------------------------

### Start SeaTunnel Engine Server Node

Source: https://seatunnel.incubator.apache.org/zh-CN/docs/2.3.13/engines/zeta/hybrid-cluster-deployment

Initializes the log directory and starts the server in daemon mode.

```bash
mkdir -p $SEATUNNEL_HOME/logs  
./bin/seatunnel-cluster.sh -d  
```

--------------------------------

### MySQL Sink Table Naming Examples

Source: https://seatunnel.incubator.apache.org/zh-CN/docs/2.3.13/connectors/sink/Jdbc

Examples of table naming configurations for MySQL sinks.

```text
1. test _${schema_name}_ ${table_name}_test
  2. sink_sinktable
  3. ss_${table_name}
```

--------------------------------

### ClickhouseFile Configuration Example

Source: https://seatunnel.incubator.apache.org/zh-CN/docs/2.3.13/connectors/sink/ClickhouseFile

Example configuration block for the ClickhouseFile sink connector.

```hocon
ClickhouseFile {  
  host = "192.168.0.1:8123"  
  database = "default"  
  table = "fake_all"  
  username = "default"  
  password = ""  
  clickhouse_local_path = "/Users/seatunnel/Tool/clickhouse local"  
  sharding_key = "age"  
  node_free_password = false  
  node_pass = [{  
    node_address = "192.168.0.1"  
    password = "seatunnel"  
  }]  
}  
```

--------------------------------

### Web3j Source Configuration Example

Source: https://seatunnel.incubator.apache.org/zh-CN/docs/2.3.13/connectors/source/Web3j

Example configuration for a batch job using the Web3j source to connect to an Ethereum node via Infura.

```hocon
env {  
  parallelism = 1  
  job.mode = "BATCH"  
}  
  
source {  
  Web3j {  
    url = "https://mainnet.infura.io/v3/xxxxx"  
  }  
}  
  
# 控制台打印读取的 Http 数据  
sink {  
  Console {  
    parallelism = 1  
  }  
}  
```

--------------------------------

### PostgreSQL Sink Table Naming Examples

Source: https://seatunnel.incubator.apache.org/zh-CN/docs/2.3.13/connectors/sink/Jdbc

Examples of table naming configurations for PostgreSQL and similar database sinks.

```text
1. ${schema_name}.${table_name}_test
  2. dbo.tt_${table_name}_sink
  3. public.sink_table
```

--------------------------------

### Set Metalake URL Example

Source: https://seatunnel.incubator.apache.org/zh-CN/docs/2.3.13/introduction/configuration/metalake

Example of a configured METALAKE_URL pointing to a specific test metalake.

```text
http://localhost:8090/api/metalakes/test_metalake/catalogs/  
```

--------------------------------

### Multi-table Iceberg Sink Examples

Source: https://seatunnel.incubator.apache.org/zh-CN/docs/2.3.13/connectors/sink/Iceberg

Examples demonstrating how to use placeholders for dynamic namespace and table names when writing to multiple tables.

```hocon
env {  
  parallelism = 1  
  job.mode = "STREAMING"  
  checkpoint.interval = 5000  
}  
  
source {  
  Mysql-CDC {  
    url = "jdbc:mysql://127.0.0.1:3306/seatunnel"  
    username = "root"  
    password = "******"  
      
    table-names = ["seatunnel.role","seatunnel.user","galileo.Bucket"]  
  }  
}  
  
transform {  
}  
  
sink {  
  Iceberg {  
    ...  
    namespace = "${database_name}_test"  
    table = "${table_name}_test"  
  }  
}  
```

```hocon
env {  
  parallelism = 1  
  job.mode = "BATCH"  
}  
  
source {  
  Jdbc {  
    driver = oracle.jdbc.driver.OracleDriver  
    url = "jdbc:oracle:thin:@localhost:1521/XE"  
    user = testUser  
    password = testPassword  
  
    table_list = [  
      {  
        table_path = "TESTSCHEMA.TABLE_1"  
      },  
      {  
        table_path = "TESTSCHEMA.TABLE_2"  
      }  
    ]  
  }  
}  
  
transform {  
}  
  
sink {  
  Iceberg {  
    ...  
    namespace = "${schema_name}_test"  
    table = "${table_name}_test"  
  }  
}  
```

--------------------------------

### Redis Data Formatting Examples

Source: https://seatunnel.incubator.apache.org/zh-CN/docs/2.3.13/connectors/sink/Redis

Examples showing how upstream data is serialized into Redis based on the configured format.

```json
{"code":  200, "data":  "获取成功", "success":  "true"}
```

```text
200#get success#true
```

--------------------------------

### Configure Third-party Jars

Source: https://seatunnel.incubator.apache.org/zh-CN/docs/2.3.13/introduction/configuration/JobEnvConfig

Example of how to load third-party jars using the jars parameter.

```text
jars="file://local/jar1.jar;file://local/jar2.jar"
```

--------------------------------

### Cluster Overview Response

Source: https://seatunnel.incubator.apache.org/zh-CN/docs/2.3.13/engines/zeta/rest-api-v1

Example JSON response for the GET /hazelcast/rest/maps/overview endpoint.

```json
{
    "projectVersion":"2.3.10-SNAPSHOT",
    "gitCommitAbbrev":"DeadD0d0",
    "totalSlot":"0",
    "unassignedSlot":"0",
    "works":"1",
    "runningJobs":"0",
    "finishedJobs":"0",
    "failedJobs":"0",
    "cancelledJobs":"0"
}
```

--------------------------------

### Connector Command Examples

Source: https://seatunnel.incubator.apache.org/zh-CN/docs/2.3.13/engines/command/connector-check

Common usage patterns for listing plugins and retrieving option rules.

```bash
# List all supported connectors(sources and sinks) and transforms  
bin/seatunnel-connector.sh -l  
# List all supported sinks  
bin/seatunnel-connector.sh -l -pt sink  
# Get option rule of the connector or transform by the name  
bin/seatunnel-connector.sh -o Paimon  
# Get option rule of paimon sink  
bin/seatunnel-connector.sh -o Paimon -pt sink
```

--------------------------------

### Thread Dump Response

Source: https://seatunnel.incubator.apache.org/zh-CN/docs/2.3.13/engines/zeta/rest-api-v1

Example JSON response structure for the GET /hazelcast/rest/maps/thread-dump endpoint.

```json
[
  {
    "threadName": "",
    "threadId": 0,
    "threadState": "",
    "stackTrace": ""
  }
]
```

--------------------------------

### Running Jobs Overview Response

Source: https://seatunnel.incubator.apache.org/zh-CN/docs/2.3.13/engines/zeta/rest-api-v1

Example JSON response structure for the GET /hazelcast/rest/maps/running-jobs endpoint.

```json
[
  {
    "jobId": "",
    "jobName": "",
    "jobStatus": "",
    "envOptions": {
    },
    "createTime": "",
    "jobDag": {
      "jobId": "",
      "envOptions": [],
      "vertexInfoMap": [
        {
          "vertexId": 1,
          "type": "",
          "vertexName": "",
          "tablePaths": [
            ""
          ]
        }
      ],
      "pipelineEdges": {}
    },
    "pluginJarsUrls": [
    ],
    "isStartWithSavePoint": false,
    "metrics": {
      "sourceReceivedCount": "",
      "sinkWriteCount": ""
    }
  }
]
```

--------------------------------

### Get Job Checkpoint Overview Response

Source: https://seatunnel.incubator.apache.org/zh-CN/docs/2.3.13/engines/zeta/rest-api-v2

Example response showing checkpoint statistics for a specific job.

```json
{
  "jobId": "1234567890",
  "updatedAt": 1720000000123,
  "pipelines": [
    {
      "pipelineId": 1,
      "counts": {
        "triggered": 10,
        "completed": 8,
        "failed": 1,
        "inProgress": 1,
        "restored": 2
      },
      "latestCompleted": {
        "checkpointId": 9,
        "checkpointType": "CHECKPOINT_TYPE",
        "status": "COMPLETED",
        "triggerTimestamp": 1720000000000,
        "completedTimestamp": 1720000000450,
        "durationMillis": 450,
        "stateSize": 128934
      },
      "latestFailed": {
        "checkpointId": 8,
        "checkpointType": "CHECKPOINT_TYPE",
        "status": "FAILED",
        "triggerTimestamp": 1719999995000,
        "failureReason": "CHECKPOINT_EXPIRED"
      },
      "latestSavepoint": null,
      "inProgress": [
        {
          "checkpointId": 10,
          "checkpointType": "CHECKPOINT_TYPE",
          "triggerTimestamp": 1720000005000,
          "acknowledged": 2,
          "total": 4
        }
      ],
      "history": [
        {
          "pipelineId": 1,
          "checkpoint": {
            "checkpointId": 9,
            "checkpointType": "CHECKPOINT_TYPE",
            "status": "COMPLETED",
            "triggerTimestamp": 1720000000000,
            "completedTimestamp": 1720000000450,
            "durationMillis": 450,
            "stateSize": 128934
          }
        }
      ]
    }
]
}
```

--------------------------------

### Hive on OSS Configuration Example

Source: https://seatunnel.incubator.apache.org/zh-CN/docs/2.3.13/connectors/source/Hive

A sample SeaTunnel configuration file demonstrating how to set up Hive source and sink for OSS.

```hocon
env {  
  parallelism = 1  
  job.mode = "BATCH"  
}  
  
source {  
  Hive {  
    table_name = "test_hive.test_hive_sink_on_oss"  
    metastore_uri = "thrift://master-1-1.c-1009b01725b501f2.cn-wulanchabu.emr.aliyuncs.com:9083"  
    hive.hadoop.conf-path = "/tmp/hadoop"  
    hive.hadoop.conf = {  
        bucket="oss://emr-osshdfs.cn-wulanchabu.oss-dls.aliyuncs.com"  
    }  
  }  
}  
  
sink {  
  Hive {  
    table_name = "test_hive.test_hive_sink_on_oss_sink"  
    metastore_uri = "thrift://master-1-1.c-1009b01725b501f2.cn-wulanchabu.emr.aliyuncs.com:9083"  
    hive.hadoop.conf-path = "/tmp/hadoop"  
    hive.hadoop.conf = {  
        bucket="oss://emr-osshdfs.cn-wulanchabu.oss-dls.aliyuncs.com"  
    }  
  }  
}  
```

--------------------------------

### Range Query Configuration

Source: https://seatunnel.incubator.apache.org/zh-CN/docs/2.3.13/connectors/source/Prometheus

Example configuration for performing a range query, requiring start, end, and step parameters.

```hocon
source {  
  Prometheus {  
    plugin_output = "http"  
    url = "http://mockserver:1080"  
    query = "up"  
    query_type = "Range"  
    content_field = "$.data.result.*"  
    format = "json"  
    start = "2024-07-22T20:10:30.781Z"  
    end = "2024-07-22T20:11:00.781Z"  
    step = "15s"  
    schema = {  
        fields {  
            metric = "map<string, string>"  
            value = double  
            time = long  
            }  
        }  
  }
}
```

--------------------------------

### Text Format Example

Source: https://seatunnel.incubator.apache.org/zh-CN/docs/2.3.13/connectors/source/Redis

Example of raw text data from Redis.

```text
200#get success#true
```

--------------------------------

### Basic Kafka Source Configuration

Source: https://seatunnel.incubator.apache.org/zh-CN/docs/2.3.13/connectors/source/Kafka

A simple example showing how to read from multiple Kafka topics and print the data to the console.

```hocon
# 定义运行环境  
env {  
  parallelism = 2  
  job.mode = "BATCH"  
}  
source {  
  Kafka {  
    schema = {  
      fields {  
        name = "string"  
        age = "int"  
      }  
    }  
    format = text  
    field_delimiter = "#"  
    topic = "topic_1,topic_2,topic_3"  
    bootstrap.servers = "localhost:9092"  
    kafka.config = {  
      client.id = client_1  
      max.poll.records = 500  
      auto.offset.reset = "earliest"  
      enable.auto.commit = "false"  
    }  
  }    
}  
sink {  
  Console {}  
}  
```

--------------------------------

### Text Data Example

Source: https://seatunnel.incubator.apache.org/zh-CN/docs/2.3.13/connectors/source/CosFile

Example of raw text data format.

```text
tyrantlucifer#26#male
```

--------------------------------

### JSON Data Format Example

Source: https://seatunnel.incubator.apache.org/zh-CN/docs/2.3.13/connectors/source/Redis

Example of raw JSON data from Redis.

```json
{"code":  200, "data":  "get success", "success":  true}
```

--------------------------------

### Simple Sls Configuration Example

Source: https://seatunnel.incubator.apache.org/zh-CN/docs/2.3.13/connectors/source/Sls

A basic configuration for a SeaTunnel job reading from an SLS logstore and printing the output to the console.

```hocon
# Defining the runtime environment  
env {  
  parallelism = 2  
  job.mode = "STREAMING"  
  checkpoint.interval = 30000  
}  
  
source {  
  Sls {  
    endpoint = "cn-hangzhou-intranet.log.aliyuncs.com"  
    project = "project1"  
    logstore = "logstore1"  
    access_key_id = "xxxxxxxxxxxxxxxxxxxxxxxx"  
    access_key_secret = "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"  
    schema = {  
      fields = {  
            id = "int"  
            name = "string"  
            description = "string"  
            weight = "string"  
      }  
    }  
  }  
}  
  
sink {  
  Console {  
  }  
}  
```

--------------------------------

### Full Job Configuration Example

Source: https://seatunnel.incubator.apache.org/zh-CN/docs/2.3.13/transforms/replace

A complete SeaTunnel job configuration including a FakeSource, the Replace transform, and a Console sink.

```hocon
env {  
  job.mode = "BATCH"  
}  
  
source {  
  FakeSource {  
    plugin_output = "fake"  
    row.num = 100  
    schema = {  
      fields {  
        id = "int"  
        name = "string"  
      }  
    }  
  }  
}  
  
transform {  
  Replace {  
    plugin_input = "fake"  
    plugin_output = "fake1"  
    replace_field = "name"  
    pattern = ".+"  
    replacement = "b"  
    is_regex = true  
  }  
}  
  
sink {  
  Console {  
    plugin_input = "fake1"  
  }  
}  
```

--------------------------------

### JSON Data Example

Source: https://seatunnel.incubator.apache.org/zh-CN/docs/2.3.13/connectors/source/Github

Example of raw JSON data returned from an API.

```json
{
  "code": 200,
  "data": "get success",
  "success": true
}
```

--------------------------------

### Simple Fluss Sink Configuration

Source: https://seatunnel.incubator.apache.org/zh-CN/docs/2.3.13/connectors/sink/Fluss

A basic configuration example showing how to set up a Fluss sink with a single table.

```hocon
env {  
  parallelism = 1  
  job.mode = "BATCH"  
}  
  
source {  
  # This is a example source plugin **only for test and demonstrate the feature source plugin**  
  FakeSource {  
    parallelism = 1  
    tables_configs = [  
        {  
        row.num = 7  
          schema {  
            table = "test.table1"  
            fields {  
            fbytes = bytes  
            fboolean = boolean  
            fint = int  
            ftinyint = tinyint  
            fsmallint = smallint  
            fbigint = bigint  
            ffloat = float  
            fdouble = double  
            fdecimal = "decimal(30, 8)"  
            fstring = string  
            fdate = date  
            ftime = time  
            ftimestamp = timestamp  
            ftimestamp_ltz = timestamp_tz  
            }  
        }  
        rows = [  
      {  
        kind = INSERT  
        fields = ["bWlJWmo=", true, 1940337748, 73, 17489, 7408919466156976747, 9.434991E37, 3.140411637757371E307, 4029933791018936061944.80602290, "aaaaa", "2025-01-03", "02:30:10", "2025-05-27T21:56:09", "2025-09-28T02:54:08+08:00"]  
      }  
      {  
        kind = INSERT  
        fields = ["bWlJWmo=", true, 90650390, 37, 22504, 5851888708829345169, 2.6221706E36, 1.8915341983748786E307, 3093109630614622831876.71725344, "bbbbb", "2025-01-01", "21:22:44", "2025-05-08T05:26:18", "2025-08-04T16:49:45+08:00"]  
      }  
      {  
        kind = INSERT  
        fields = ["bWlJWmo=", true, 2146418323, 79, 19821, 6393905306944584839, 2.0462337E38, 1.4868114385836557E308, 5594947262031769994080.35717665, "ccccc", "2025-10-06", "22:10:40", "2025-03-25T01:49:14", "2025-07-03T11:52:06+08:00"]  
      }  
      {  
        kind = DELETE  
        fields = ["bWlJWmo=", true, 2146418323, 79, 19821, 6393905306944584839, 2.0462337E38, 1.4868114385836557E308, 5594947262031769994080.35717665, "ccccc", "2025-10-06", "22:10:40", "2025-03-25T01:49:14", "2025-07-03T11:52:06+08:00"]  
      }  
      {  
        kind = INSERT  
        fields = ["bWlJWmo=", true, 82794384, 27, 30339, 5826566947079347516, 2.2137477E37, 1.7737681870839753E308, 3984670873242882274814.90739768, "ddddd", "2025-09-13", "10:32:52", "2025-01-27T19:20:51", "2025-11-07T02:38:54+08:00"]  
      }  
      {  
        kind = UPDATE_BEFORE  
        fields = ["bWlJWmo=", true, 82794384, 27, 30339, 5826566947079347516, 2.2137477E37, 1.7737681870839753E308, 3984670873242882274814.90739768, "ddddd", "2025-09-13", "10:32:52", "2025-01-27T19:20:51", "2025-11-07T02:38:54+08:00"]  
      }  
      {  
        kind = UPDATE_AFTER  
        fields = ["bWlJWmo=", true, 388742243, 89, 15831, 159071788675312856, 7.310445E37, 1.2166972324288247E308, 7994947075691901110245.55960937, "ddddd", "2025-01-04", "15:28:07", "2025-07-18T08:59:49", "2025-09-12T23:46:25+08:00"]  
      }  
    ]  
    }  
      ]  
}  
}  
  
transform {  
}  
  
sink {  
  Fluss {  
    bootstrap.servers="fluss_coordinator_e2e:9123"  
    database = "fluss_db_${database_name}"  
    table = "fluss_tb_${table_name}"  
  }  
}  
```

--------------------------------

### JSON Configuration Format

Source: https://seatunnel.incubator.apache.org/zh-CN/docs/2.3.13/introduction/concepts/config

Example of a complete SeaTunnel configuration file using the JSON format.

```json
  
{  
  "env": {  
    "job.mode": "batch"  
  },  
  "source": [  
    {  
      "plugin_name": "FakeSource",  
      "plugin_output": "fake",  
      "row.num": 100,  
      "schema": {  
        "fields": {  
          "name": "string",  
          "age": "int",  
          "card": "int"  
        }  
      }  
    }  
  ],  
  "transform": [  
    {  
      "plugin_name": "Filter",  
      "plugin_input": "fake",  
      "plugin_output": "fake1",  
      "fields": ["name", "card"]  
    }  
  ],  
  "sink": [  
    {  
      "plugin_name": "Clickhouse",  
      "host": "clickhouse:8123",  
      "database": "default",  
      "table": "seatunnel_console",  
      "fields": ["name", "card"],  
      "username": "default",  
      "password": "",  
      "plugin_input": "fake1"  
    }  
  ]  
}  
  
```

--------------------------------

### Hash Key Example Data

Source: https://seatunnel.incubator.apache.org/zh-CN/docs/2.3.13/connectors/source/Redis

Example structure of a hash key value in Redis.

```json
{
  "001": {
    "name": "tyrantlucifer",
    "age": 26
  },
  "002": {
    "name": "Zongwen",
    "age": 26
  }
}
```

--------------------------------

### IoTDB Query Example

Source: https://seatunnel.incubator.apache.org/zh-CN/docs/2.3.13/connectors/source/IoTDB

Example SQL query used to retrieve data from IoTDB.

```sql
IoTDB> SELECT temperature, moisture, c_int, c_bigint, c_float, c_double, c_string, c_boolean FROM root.test_group.* WHERE time < 4102329600000 align by device;  
+------------------------+------------------------+--------------+-----------+--------+--------------+----------+---------+---------+----------+  
|                    Time|                  Device|   temperature|   moisture|   c_int|      c_bigint|   c_float| c_double| c_string| c_boolean|  
+------------------------+------------------------+--------------+-----------+--------+--------------+----------+---------+---------+----------+  
|2022-09-25T00:00:00.001Z|root.test_group.device_a|          36.1|        100|       1|   21474836470|      1.0f|     1.0d|      abc|      true|  
|2022-09-25T00:00:00.001Z|root.test_group.device_b|          36.2|        101|       2|   21474836470|      2.0f|     2.0d|      abc|      true|  
|2022-09-25T00:00:00.001Z|root.test_group.device_c|          36.3|        102|       3|   21474836470|      3.0f|     3.0d|      abc|      true|  
+------------------------+------------------------+--------------+-----------+--------+--------------+----------+---------+---------+----------+  
```

--------------------------------

### Milvus Source Configuration Example

Source: https://seatunnel.incubator.apache.org/zh-CN/docs/2.3.13/connectors/source/Milvus

Basic configuration for the Milvus source connector, requiring the URL, authentication token, and database name.

```hocon
source {  
  Milvus {  
    url = "http://127.0.0.1:19530"  
    token = "username:password"  
    database = "default"  
  }  
}  
```

--------------------------------

### JSON Data Format Examples

Source: https://seatunnel.incubator.apache.org/zh-CN/docs/2.3.13/connectors/source/SftpFile

Examples of JSON data structures and the corresponding schema configuration.

```json
{"code":  200, "data":  "get success", "success":  true}
```

```json
{"code":  200, "data":  "get success", "success":  true}
{"code":  300, "data":  "get failed", "success":  false}
```

```hocon
schema {  
    fields {  
        code = int  
        data = string  
        success = boolean  
    }  
}  
```

--------------------------------

### InfluxDB SQL Query Example

Source: https://seatunnel.incubator.apache.org/zh-CN/docs/2.3.13/connectors/source/InfluxDB

Example of a SQL query used to fetch data from InfluxDB.

```sql
select name,age from test
```

--------------------------------

### 创建Docker网络

Source: https://seatunnel.incubator.apache.org/zh-CN/docs/2.3.13/getting-started/docker

为SeaTunnel集群创建一个专用的Docker网络。

```bash
docker network create seatunnel-network  
```

--------------------------------

### File Structure Example

Source: https://seatunnel.incubator.apache.org/zh-CN/docs/2.3.13/connectors/source/CosFile

Example file structure used for demonstrating file filtering patterns.

```text
/data/seatunnel/20241001/report.txt  
/data/seatunnel/20241007/abch202410.csv  
/data/seatunnel/20241002/abcg202410.csv  
/data/seatunnel/20241005/old_data.csv  
/data/seatunnel/20241012/logo.png  
```

--------------------------------

### Configure Job Mode

Source: https://seatunnel.incubator.apache.org/zh-CN/docs/2.3.13/introduction/configuration/JobEnvConfig

Examples of setting the job mode to either BATCH or STREAMING.

```text
job.mode = "BATCH"
```

```text
job.mode = "STREAMING"
```

--------------------------------

### Simple Linear Pipeline Configuration

Source: https://seatunnel.incubator.apache.org/zh-CN/docs/2.3.13/architecture/engine/dag-execution

A basic configuration example resulting in a single execution pipeline.

```text
source { JDBC { } }  
transform { Sql { } }  
sink { Elasticsearch { } }  
```

```text
流水线 1: [JDBC 数据源] → [SQL 转换器] → [Elasticsearch 目标端]  
```

--------------------------------

### 运行SeaTunnel本地模式任务

Source: https://seatunnel.incubator.apache.org/zh-CN/docs/2.3.13/getting-started/docker

通过docker run命令启动容器并执行任务，支持挂载本地配置文件及设置JVM选项。

```bash
# Run fake source to console sink  
docker run --rm -it apache/seatunnel:<version_tag> ./bin/seatunnel.sh -m local -c config/v2.batch.config.template  
  
# Run job with custom config file  
docker run --rm -it -v /<The-Config-Directory-To-Mount>/:/config apache/seatunnel:<version_tag> ./bin/seatunnel.sh -m local -c /config/fake_to_console.conf  
  
# Example  
# If you config file is in /tmp/job/fake_to_console.conf  
docker run --rm -it -v /tmp/job/:/config apache/seatunnel:<version_tag> ./bin/seatunnel.sh -m local -c /config/fake_to_console.conf  
  
# Set JVM options when running  
docker run --rm -it -v /tmp/job/:/config apache/seatunnel:<version_tag> ./bin/seatunnel.sh -DJvmOption="-Xms4G -Xmx4G" -m local -c /config/fake_to_console.conf
```

--------------------------------

### 配置 SEATUNNEL_HOME 环境变量

Source: https://seatunnel.incubator.apache.org/zh-CN/docs/2.3.13/engines/zeta/hybrid-cluster-deployment

通过创建 profile 文件设置 SEATUNNEL_HOME 路径并将其添加到系统 PATH 中。

```bash
export SEATUNNEL_HOME=${seatunnel install path}  
export PATH=$PATH:$SEATUNNEL_HOME/bin  
```

--------------------------------

### JSON Content Extraction Example

Source: https://seatunnel.incubator.apache.org/zh-CN/docs/2.3.13/connectors/source/Github

Example of nested JSON data structure for content field extraction.

```json
{
  "store": {
    "book": [
      {
        "category": "reference",
        "author": "Nigel Rees",
        "title": "Sayings of the Century",
        "price": 8.95
      },
      {
        "category": "fiction",
        "author": "Evelyn Waugh",
        "title": "Sword of Honour",
        "price": 12.99
      }
    ],
    "bicycle": {
      "color": "red",
      "price": 19.95
    }
  },
  "expensive": 10
}
```

--------------------------------

### JSON Data Format Examples

Source: https://seatunnel.incubator.apache.org/zh-CN/docs/2.3.13/connectors/source/CosFile

Examples of single and multi-line JSON data structures for the CosFile connector.

```json
{"code":  200, "data":  "get success", "success":  true}
```

```json
{"code":  200, "data":  "get success", "success":  true}
{"code":  300, "data":  "get failed", "success":  false}
```

--------------------------------

### 配置 JVM 参数

Source: https://seatunnel.incubator.apache.org/zh-CN/docs/2.3.13/engines/zeta/user-command

通过配置文件或命令行参数为客户端设置 JVM 选项。

```bash
sh bin/seatunnel.sh --config $SEATUNNEL_HOME/config/v2.batch.config.template -DJvmOption="-Xms2G -Xmx2G"
```

--------------------------------

### Define Source Schema

Source: https://seatunnel.incubator.apache.org/zh-CN/docs/2.3.13/connectors/sink/IoTDBv2

Example schema definition for the FakeSource used in the following IoTDB sink examples.

```hocon
env {  
  parallelism = 2  
  job.mode = "BATCH"  
}  
  
source {  
  FakeSource {  
    ...  
    schema = {  
      fields {  
        ts = timestamp  
        model_id = string  
        region = string  
        tag = string  
        status = boolean  
        arrival_date = date  
        temperature = double  
      }  
    }  
  }  
}  
```

--------------------------------

### IoTDB Time-Based Partitioning Logic

Source: https://seatunnel.incubator.apache.org/zh-CN/docs/2.3.13/connectors/source/IoTDB

Example illustrating how the connector splits time ranges into partitions based on lower_bound, upper_bound, and num_partitions settings.

```text
     将时间范围分割成 numPartitions 个分区  
       
     若 numPartitions = 1，使用完整的时间范围  
     若 numPartitions < (upper_bound - lower_bound)，使用 (upper_bound - lower_bound) 个分区  
       
     例：lower_bound = 1, upper_bound = 10, numPartitions = 2  
         sql = "select * from test where age > 0 and age < 10"  
       
     分区结果：  
     split 1: select * from test  where (time >= 1 and time < 6)  and (  age > 0 and age < 10 )  
     split 2: select * from test  where (time >= 6 and time < 11) and (  age > 0 and age < 10 )  
```

--------------------------------

### Batch Write Query Example

Source: https://seatunnel.incubator.apache.org/zh-CN/docs/2.3.13/connectors/sink/Neo4j

Example of a Cypher query using the UNWIND clause for batch processing.

```cypher
unwind $ttt as row create (n:Label) set n.name = row.name,n.age = rw.age  
```

--------------------------------

### Prepare Hive Connector Environment

Source: https://seatunnel.incubator.apache.org/zh-CN/docs/2.3.13/connectors/source/Hive

Commands to create the required library directory and download the necessary Hive execution JAR file.

```bash
mkdir -p ${SEATUNNEL_HOME}/plugins/Hive/lib
```

```bash
cd ${SEATUNNEL_HOME}/plugins/Hive/lib
wget https://repo1.maven.org/maven2/org/apache/hive/hive-exec/2.3.9/hive-exec-2.3.9.jar
```

--------------------------------

### Configure Schema Save Mode

Source: https://seatunnel.incubator.apache.org/zh-CN/docs/2.3.13/connectors/sink/Databend

Example demonstrating how to set schema and data save modes for table initialization.

```hocon
sink {  
  Databend {  
    url = "jdbc:databend://localhost:8000"  
    username = "root"  
    password = ""  
    database = "default"  
    table = "target_table"  
    schema_save_mode = "RECREATE_SCHEMA"  
    data_save_mode = "APPEND_DATA"  
  }  
}  
```

--------------------------------

### MongoDB Connection URI Examples

Source: https://seatunnel.incubator.apache.org/zh-CN/docs/2.3.13/connectors/sink/MongoDB

Various connection string formats for different MongoDB deployment topologies.

```text
mongodb://127.0.0.0:27017/mydb  
```

```text
mongodb://127.0.0.0:27017/mydb?replicaSet=xxx  
```

```text
mongodb://admin:password@127.0.0.0:27017/mydb?replicaSet=xxx&authSource=admin  
```

```text
mongodb://127.0.0.1:27017,127.0.0.2:27017,127.0.0.3:27017/mydb?replicaSet=xxx  
```

```text
mongodb://127.0.0.0:27017/mydb  
```

```text
mongodb://192.168.0.1:27017,192.168.0.2:27017,192.168.0.3:27017/mydb  
```

--------------------------------

### File Filter Pattern Examples

Source: https://seatunnel.incubator.apache.org/zh-CN/docs/2.3.13/connectors/source/FtpFile

Examples of regular expressions used for filtering files based on name or path structure.

```text
/data/seatunnel/20241001/report.txt  
/data/seatunnel/20241007/abch202410.csv  
/data/seatunnel/20241002/abcg202410.csv  
/data/seatunnel/20241005/old_data.csv  
/data/seatunnel/20241012/logo.png  
```

```text
.*.txt  
```

```text
/data/seatunnel/20241001/report.txt  
```

```text
abc.*  
```

```text
/data/seatunnel/20241007/abch202410.csv  
/data/seatunnel/20241002/abcg202410.csv  
```

```text
/data/seatunnel/20241007/abc[h,g].*  
```

```text
/data/seatunnel/20241007/abch202410.csv  
```

```text
/data/seatunnel/202410\d*/.*.csv  
```

```text
/data/seatunnel/20241007/abch202410.csv  
/data/seatunnel/20241002/abcg202410.csv  
/data/seatunnel/20241005/old_data.csv  
```

--------------------------------

### 编译与打包源码

Source: https://seatunnel.incubator.apache.org/zh-CN/docs/2.3.13/developer/setup

使用 Maven 对整个项目进行编译和打包。

```bash
mvn clean package -pl seatunnel-dist -am -Dmaven.test.skip=true
```

--------------------------------

### Multi-table InfluxDB Sink Example

Source: https://seatunnel.incubator.apache.org/zh-CN/docs/2.3.13/connectors/sink/InfluxDB

Example demonstrating how to use the InfluxDB sink with multiple tables from a MySQL CDC source.

```hocon
env {  
  parallelism = 1  
  job.mode = "STREAMING"  
  checkpoint.interval = 5000  
}  
  
source {  
  Mysql-CDC {  
    url = "jdbc:mysql://127.0.0.1:3306/seatunnel"  
    username = "root"  
    password = "******"  
      
    table-names = ["seatunnel.role","seatunnel.user","galileo.Bucket"]  
  }  
}  
  
transform {  
}  
  
sink {  
  InfluxDB {  
    url = "http://influxdb-host:8086"  
    database = "test"  
    measurement = "${table_name}_test"  
  }  
}  
```
Tokens:
Raw