### Simple Sls Source Example

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

Reads data from an Sls logstore and prints it to the console. Ensure SeaTunnel is installed and configured.

```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 {
  }
}

```

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

### Simple Sls Sink Example

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

This example demonstrates writing data to an Aliyun Sls logstore. Ensure SeaTunnel is installed and configured before running this job. It includes runtime environment settings and a FakeSource for generating data.

```hocon
env {
  parallelism = 2
  job.mode = "STREAMING"
  checkpoint.interval = 30000
}
source {
  FakeSource {
    row.num = 10
    map.size = 10
    array.size = 10
    bytes.length = 10
    string.length = 10
    schema = {
      fields = {
            id = "int"
            name = "string"
            description = "string"
            weight = "string"
      }
    }
  }
}

sink {
  Sls {
    endpoint = "cn-hangzhou-intranet.log.aliyuncs.com"
    project = "project1"
    logstore = "logstore1"
    access_key_id = "xxxxxxxxxxxxxxxxxxxxxxxx"
    access_key_secret = "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
  }
}

```

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

### Install Connectors

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

Execute the installation script to install the configured connectors. This command should be run from the SeaTunnel home directory.

```bash
# Install connectors
sh bin/install-plugin.sh

```

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

### Start Minikube Cluster

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

Starts a local Kubernetes cluster using Minikube. Ensure you have Minikube installed and configured.

```bash
minikube start --kubernetes-version=v1.23.3  

```

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

### Simple Example with FakeSource

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

This example demonstrates a basic SeaTunnel job that generates data using FakeSource and writes it to an S3 file. Ensure the S3 path exists and SeaTunnel is installed and configured.

```hocon
env {
  parallelism = 1
  job.mode = "BATCH"
}

source {
  FakeSource {
    parallelism = 1
    plugin_output = "fake"
    row.num = 16
    schema = {
      fields {
        c_map = "map<string, array<int>>"
        c_array = "array<int>"
        name = string
        c_boolean = boolean
        age = tinyint
        c_smallint = smallint
        c_int = int
        c_bigint = bigint
        c_float = float
        c_double = double
        c_decimal = "decimal(16, 1)"
        c_null = "null"
        c_bytes = bytes
        c_date = date
        c_timestamp = timestamp
      }
    }
  }
}
```

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

### Install Connector Plugins

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

Execute the install-plugin.sh script to download and install connector plugins for the specified SeaTunnel version.

```bash
sh bin/install-plugin.sh 2.3.13  
```

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

### Simple Paimon Source Example

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

A basic example demonstrating how to configure the Paimon source connector with warehouse, database, and table information.

```java
source {
  Paimon {
      warehouse = "/tmp/paimon"
      database = "default"
      table = "st_test"
    }
}

```

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

### Generate Configuration Example

Source: https://seatunnel.incubator.apache.org/zh-CN/docs/2.3.13/tools/seatunnel-skill

Generate a task configuration example using the SeaTunnel Skill. For example, request a configuration for a MySQL to Elasticsearch task.

```bash
/seatunnel-skill "创建一个 MySQL 到 Elasticsearch 的任务配置"
```

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

### Simple Example Output

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

This is the expected output after successfully running a simple SeaTunnel engine example. It shows job statistics like start time, end time, and total read/write counts.

```log
2024-08-10 11:45:32,839 INFO  org.apache.seatunnel.core.starter.seatunnel.command.ClientExecuteCommand -   
***********************************************  
           Job Statistic Information  
***********************************************  
Start Time                : 2024-08-10 11:45:30  
End Time                  : 2024-08-10 11:45:32  
Total Time(s)             :                   2  
Total Read Count          :                   5  
Total Write Count         :                   5  
Total Failed Count        :                   0  
***********************************************  

```

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

### Multi-Source/Sink Configuration Example

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

Example of a SeaTunnel configuration defining multiple sources, a transformation, and multiple sinks.

```yaml
# 多数据源/Sink 配置  
env { ... }  
  
source {  
  MySQL-CDC { table = "orders" }  
  Kafka { topic = "events" }  
}  

transform {  
  Sql { query = "SELECT * FROM orders JOIN events ON ..." }  
}  

sink {  
  Elasticsearch { index = "orders" }  
  JDBC { table = "events" }  
}  

```

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

### RocketMQ Offsets Configuration Example

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

Example configuration for `start.mode.offsets` when consuming from specific offsets in RocketMQ.

```plaintext
start.mode.offsets = {
  topic1-0 = 70
  topic1-1 = 10
  topic1-2 = 10
}
```

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

### Cassandra Sink Configuration Example

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

This example demonstrates how to configure the Cassandra sink connector with basic connection details and authentication.

```hocon
sink {
 Cassandra {
     host = "localhost:9042"
     username = "cassandra"
     password = "cassandra"
     datacenter = "datacenter1"
     keyspace = "test"
    }
}

```

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

### Install SeaTunnel Helm Chart

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

Installs the SeaTunnel Helm chart with default configurations. Remember to choose the appropriate version.

```bash
# Choose the corresponding version yourself  
export VERSION=2.3.10  
helm pull oci://registry-1.docker.io/apache/seatunnel-helm --version ${VERSION}  
tar -xvf seatunnel-helm-${VERSION}.tgz  
cd seatunnel-helm  
helm install seatunnel .  

```

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

### Example: Run Flink 15 Batch Job

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

Example of running a Flink 15 batch job with a specified configuration file.

```bash
bin/start-seatunnel-flink-15-connector-v2.sh --config config/v2.batch.config.template  
```

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

### Doris Sink Configuration Example

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

Example configuration for the Doris sink connector, specifying connection details and data handling options.

```toml
sink {
    name = "doris"
    enable_2pc = true
    fenodes = "192.168.1.100:8030"
    username = "root"
    password = "123456"
    database = "test_db"
    table = "test_table"
    sink.label-prefix = "doris_label"
    doris.batch.size = 1024
    sink.enable-delete = true
    schema_save_mode = "CREATE_SCHEMA_WHEN_NOT_EXIST"
    data_save_mode = "APPEND_DATA"
}
```

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

### Example Configuration with PostgreSQL and Console Sink

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

This configuration example demonstrates using PostgreSQL as a data source and outputting to the console. Remember to adjust the `pom.xml` for connector dependencies if using non-default connectors.

```hocon
env {
  parallelism = 1
  job.mode = "BATCH"
}

source {
    Jdbc {
        driver = org.postgresql.Driver
        url = "jdbc:postgresql://host:port/database"
        user = "postgres"
        password = "123456"
        query = "select * from test"
        table_path = "database.test"
    }
}

sink {
  Console {}
}

```

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

### Install Cert-Manager for Webhook

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

Install the cert-manager to enable webhook components in your Kubernetes cluster. This is a one-time setup per cluster.

```bash
kubectl create -f https://github.com/jetstack/cert-manager/releases/download/v1.8.2/cert-manager.yaml  

```

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

### Start SeaTunnel Flink Job (Flink 1.15.x - 1.18.x)

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

Command to start a SeaTunnel application using Flink versions 1.15.x to 1.18.x. Ensure you are in the SeaTunnel installation directory.

```bash
cd "apache-seatunnel-${version}"
./bin/start-seatunnel-flink-15-connector-v2.sh --config ./config/v2.streaming.conf.template
```

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

### Example: Run Local Mode Job with Custom Config

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

An example demonstrating how to run a local mode job with a custom configuration file located in `/tmp/job/`.

```bash
# 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
```

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

### Start SeaTunnel Flink Job (Flink 1.12.x - 1.14.x)

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

Command to start a SeaTunnel application using Flink versions 1.12.x to 1.14.x. Ensure you are in the SeaTunnel installation directory.

```bash
cd "apache-seatunnel-${version}"
./bin/start-seatunnel-flink-13-connector-v2.sh --config ./config/v2.streaming.conf.template
```

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

### OssFile Source Configuration Example

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

Example configuration for the OssFile source connector, demonstrating how to specify file paths, credentials, format types, and filtering patterns. Use this to read data from OSS buckets.

```hocon
env {
  parallelism = 1
  job.mode = "BATCH"
}

source {
  OssFile {
    path = "/seatunnel/orc"
    bucket = "oss://tyrantlucifer-image-bed"
    access_key = "xxxxxxxxxxxxxxxxx"
    access_secret = "xxxxxxxxxxxxxxxxxxxxxx"
    endpoint = "oss-cn-beijing.aliyuncs.com"
    file_format_type = "orc"
    // 文件示例 abcD2024.csv
    file_filter_pattern = "abc[DX]*.*"
    // 筛选最后修改日期在 20240101 和 20240105 (不包括该日期)
    file_filter_modified_start = "2024-01-01 00:00:00"
    file_filter_modified_end = "2024-01-05 00:00:00"
  }
}

sink {
  Console {
  }
}

```

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

### CosFile File Filter Pattern Example 2

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

This example demonstrates matching all files starting with 'abc' using a regular expression. The output shows the matched file paths.

```regex
abc.*
```

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

### CosFile File Filter Pattern Example 4

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

This example demonstrates matching files in a third-level directory starting with '202410' and ending with '.csv'. The output shows the matched file paths.

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

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

### Complete SQL Configuration Example

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

This snippet shows a full SQL configuration file, including environment settings, source table definition (JDBC), sink table definition (JDBC), and an INSERT statement.

```sql
/* config  
env {  
  parallelism = 1  
  job.mode = "BATCH"  
}
*/  
  
CREATE TABLE source_table WITH (  
  'connector'='jdbc',
  'type'='source',
  'url' = 'jdbc:mysql://localhost:3306/seatunnel',
  'driver' = 'com.mysql.cj.jdbc.Driver',
  'user' = 'root',
  'password' = '123456',
  'query' = 'select * from source',
  'properties'= '{
    useSSL = false,
    rewriteBatchedStatements = true
  }'
);
  
CREATE TABLE sink_table WITH (  
  'connector'='jdbc',
  'type'='sink',
  'url' = 'jdbc:mysql://localhost:3306/seatunnel',
  'driver' = 'com.mysql.cj.jdbc.Driver',
  'user' = 'root',
  'password' = '123456',
  'generate_sink_sql' = 'true',
  'database' = 'seatunnel',
  'table' = 'sink'
);

INSERT INTO sink_table SELECT id, name, age, email FROM source_table;
```

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

### Multi-Source Console Sink Example

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

This example shows how to configure multiple data sources to write to specific console sinks by specifying the plugin_input.

```hocon
env {
  parallelism = 1
  job.mode = "STREAMING"
}

source {
  FakeSource {
    plugin_output = "fake1"
    schema = {
      fields {
        id = "int"
        name = "string"
        age = "int"
        sex = "string"
      }
    }
  }
   FakeSource {
    plugin_output = "fake2"
    schema = {
      fields {
        name = "string"
        age = "int"
      }
    }
  }
}

sink {
  Console {
    plugin_input = "fake1"
  }
  Console {
    plugin_input = "fake2"
  }
}

```

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

### FTP File Source Connector Configuration Example

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

This is a basic configuration example for the FtpFile source connector. Ensure all required parameters like host, port, user, and password are provided.

```hocon
ftp {
    host = "your_ftp_host"
    port = 21
    user = "your_username"
    password = "your_password"
    path = "/path/to/your/files"
    file_format_type = "text" # or csv, json, parquet, etc.
}
```

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

### CosFile File Filter Pattern Example 3

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

This example demonstrates matching files in a specific directory ('/data/seatunnel/20241007/') that start with 'abc' and have 'h' or 'g' as the fourth character. The output shows the matched file path.

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

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

### Persistiq Source Configuration Example

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

Example of how to configure the Persistiq source connector to read user data from a specified URL. Includes schema definition for the expected data structure.

```hocon
source {
  Persistiq{
    url = "https://api.persistiq.com/v1/users"
    password = "Your password"
    content_field = "$.users.*"
    schema = {
        fields {
          id = string
          name = string
          email = string
          activated = boolean
          default_mailbox_id = string
          salesforce_id = string
        }
    }
  }
}

```

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

### Fluss Sink Configuration Example

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

This example demonstrates a basic configuration for the Fluss sink connector, specifying bootstrap servers, database, and table names with variable substitution.

```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}"
  }
}

```

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

### REGEXP_SUBSTR Function Examples

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

Returns a substring that matches a regular expression. Supports specifying start position, occurrence, flags, and capture groups.

```sql
REGEXP_SUBSTR('2020-10-01', '\d{4}')
```

```sql
REGEXP_SUBSTR('2020-10-01', '(\d{4})-(\d{2})-(\d{2})', 1, 1, NULL, 2)
```

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

### MongoDB Source to Console Sink Job

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

Example of a SeaTunnel job that reads data from MongoDB and prints it to the console. Ensure the MongoDB connector is installed.

```hocon
# Set the basic configuration for the task to be executed
env {
  parallelism = 1
  job.mode = "BATCH"
}

# Create MongoDB source
source {
  MongoDB {
    uri = "mongodb://user:password@127.0.0.1:27017"
    database = "test_db"
    collection = "source_table"
    schema = {
      fields {
        c_map = "map<string, string>"
        c_array = "array<int>"
        c_string = string
        c_boolean = boolean
        c_int = int
        c_bigint = bigint
        c_double = double
        c_bytes = bytes
        c_date = date
        c_decimal = "decimal(38, 18)"
        c_timestamp = timestamp
        c_row = {
          c_map = "map<string, string>"
          c_array = "array<int>"
          c_string = string
          c_boolean = boolean
          c_int = int
          c_bigint = bigint
          c_double = double
          c_bytes = bytes
          c_date = date
          c_decimal = "decimal(38, 18)"
          c_timestamp = timestamp
        }
      }
    }
  }
}

# Print the read Mongodb data to the console
sink {
  Console {
    parallelism = 1
  }
}

```

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

### Set Flink Checkpointing Configuration

Source: https://seatunnel.incubator.apache.org/zh-CN/docs/2.3.13/engines/flink

Configure precise checkpointing for your Flink job by starting Flink configurations with 'flink.'. This example enables unaligned checkpointing.

```hocon
env {
  parallelism = 1
  flink.execution.checkpointing.unaligned.enabled=true
}

```

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

### Http Source Configuration Example

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

This snippet shows a complete configuration for the Http source connector in batch mode, including schema definition and sink to console.

```hocon
env {
  parallelism = 1
  job.mode = "BATCH"
}

source {
  Http {
    plugin_output = "http"
    url = "http://mockserver:1080/example/http"
    method = "GET"
    format = "json"
    schema = {
      fields {
        c_map = "map<string, string>"
        c_array = "array<int>"
        c_string = string
        c_boolean = boolean
        c_tinyint = tinyint
        c_smallint = smallint
        c_int = int
        c_bigint = bigint
        c_float = float
        c_double = double
        c_bytes = bytes
        c_date = date
        c_decimal = "decimal(38, 18)"
        c_timestamp = timestamp
        c_row = {
          C_MAP = "map<string, string>"
          C_ARRAY = "array<int>"
          C_STRING = string
          C_BOOLEAN = boolean
          C_TINYINT = tinyint
          C_SMALLINT = smallint
          C_INT = int
          C_BIGINT = bigint
          C_FLOAT = float
          C_DOUBLE = double
          C_BYTES = bytes
          C_DATE = date
          C_DECIMAL = "decimal(38, 18)"
          C_TIMESTAMP = timestamp
        }
      }
    }
  }
}

# 控制台打印读取的 Http 数据
sink {
  Console {
    parallelism = 1
  }
}

```

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

### Run SeaTunnel Application in Local Mode

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

Command to start a SeaTunnel application in local mode using a specified configuration file. Ensure you are in the SeaTunnel installation directory.

```bash
cd "apache-seatunnel-${version}"
./bin/seatunnel.sh --config ./config/v2.batch.config.template -m local
```

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

### Example of SeaTunnel Engine Local Mode

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

This Java code snippet demonstrates how to start the SeaTunnel engine in local mode, which is useful for understanding and debugging the source code.

```java
public class SeaTunnelEngineLocalExample {
    public static void main(String[] args) throws Exception {
        // Example code for running SeaTunnel in local mode
        // ...
    }
}
```

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

### Scale Cluster by Adding a Master Node

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

This command appears to be for starting a master node, which might be used in a scaling scenario or for re-initializing the master. Ensure the ST_DOCKER_MEMBER_LIST is correctly set if this is part of a distributed setup.

```bash
# 将ST_DOCKER_MEMBER_LIST设置为已经启动的master容器的ip
docker run -d --name seatunnel_master \
    --network seatunnel-network \
    --rm \
    -e ST_DOCKER_MEMBER_LIST=172.18.0.2:5801 \
    apache/seatunnel \
    ./bin/seatunnel-cluster.sh -r master
```

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

### Seatunnel Job with FakeSource and Hive Sink (Kerberos)

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

A complete Seatunnel job configuration using FakeSource to generate data and a Hive sink with Kerberos authentication. This example demonstrates a batch job setup.

```hocon
env {
  parallelism = 1
  job.mode = "BATCH"
}

source {
  FakeSource {
    schema = {
      fields {
        pk_id = bigint
        name = string
        score = int
      }
      primaryKey {
        name = "pk_id"
        columnNames = [pk_id]
      }
    }
    rows = [
      {
        kind = INSERT
        fields = [1, "A", 100]
      },
      {
        kind = INSERT
        fields = [2, "B", 100]
      },
      {
        kind = INSERT
        fields = [3, "C", 100]
      }
    ]
  }
}

sink {
  Hive {
    table_name = "default.test_hive_sink_on_hdfs_with_kerberos"
    metastore_uri = "thrift://metastore:9083"
    hive_site_path = "/tmp/hive-site.xml"
    kerberos_principal = "hive/metastore.seatunnel@EXAMPLE.COM"
    kerberos_keytab_path = "/tmp/hive.keytab"
    krb5_path = "/tmp/krb5.conf"
  }
}


```

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

### Redis Hash Key Parsing Example (All Mode)

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

Demonstrates how hash keys are parsed when hash_key_parse_mode is set to 'all'. Requires schema configuration.

```text
schema {
  fields {
    001 {
      name = string
      age = int
    }
    002 {
      name = string
      age = int
    }
  }
}

```

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

### Configuration to LogicalDag Example

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

Demonstrates how a HOCON configuration is translated into a LogicalDag representation, showing the linear flow of data from source to sink.

```hocon
env {
  parallelism = 4
}

source {
  JDBC {
    url = "jdbc:mysql://..."
    query = "SELECT * FROM orders"
  }
}

transform {
  Sql {
    query = "SELECT order_id, SUM(amount) FROM this GROUP BY order_id"
  }
}

sink {
  Elasticsearch {
    hosts = ["es-host:9200"]
    index = "orders_summary"
  }
}

```

```text
生成的 LogicalDag:

Vertex 1 (JDBC 数据源, parallelism=4)  
    │  
    ▼  
Vertex 2 (SQL 转换器, parallelism=4)  
    │  
    ▼  
Vertex 3 (Elasticsearch 目标端, parallelism=4)  


```

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

### Calculate Replica Number Based on Write Rate

Source: https://seatunnel.incubator.apache.org/zh-CN/docs/2.3.13/architecture/features/multi-table

Formula and example for calculating the required number of replicas for a table based on its write rate and the throughput of a single writer. Use this to determine an appropriate starting point for `multi_table_sink_replica`.

```text
replicaNum = ceil(表写入速率 / 单个写入器吞吐量)  
  
示例:  
  orders: 10,000 写入/秒  
  单个写入器: 2,500 写入/秒  
  replicaNum = ceil(10,000 / 2,500) = 4  
```

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

### Configuring Jdbc Source with Metalake Placeholders

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

Example of a Jdbc source configuration in SeaTunnel using Metalake placeholders for credentials. The 'sourceId' refers to the catalog name in Metalake. Placeholders must start with '${' and end with '}'. Each placeholder can only contain one such variable.

```hocon
source {
    Jdbc {
        url = "jdbc:mysql://mysql-e2e:3306/seatunnel?useSSL=false&serverTimezone=UTC&allowPublicKeyRetrieval=true"
        driver = "${jdbc-driver}"
        connection_check_timeout_sec = 100
        sourceId = "test_catalog"
        user = "${jdbc-user}"
        password = "${jdbc-password}"
        query = "select * from source"
    }
}
```

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

### FTP File Source Configuration Example

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

This is a comprehensive example of configuring the FtpFile source connector, specifying connection details, file format, schema, and delimiters.

```java
FtpFile {
  path = "/tmp/seatunnel/sink/text"
  host = "192.168.31.48"
  port = 21
  user = tyrantlucifer
  password = tianchao
  file_format_type = "text"
  schema = {
    name = string
    age = int
  }
  field_delimiter = "#"
}
```

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

### Phoenix Thin Client Driver Example

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

Example configuration for using the Phoenix thin client driver. This connects to the Phoenix QueryServer. Verify the URL and serialization format.

```java
Jdbc {
    driver = org.apache.phoenix.queryserver.client.Driver
    url = "jdbc:phoenix:thin:url=http://spark_e2e_phoenix_sink:8765;serialization=PROTOBUF"
    query = "select age, name from test.source"
}
```

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

### Simple MySQL Sink Example

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

A basic example demonstrating how to configure the MySQL Sink connector to write data from a FakeSource to a MySQL table. Ensure the MySQL table is created before running.

```hocon
env {
  parallelism.default = 1
  execution.parallelism.default = 1
}

source {
  FakeSource {
    row.num = 16
    schema.0.name = "name"
    schema.0.type = "string"
    schema.1.name = "age"
    schema.1.type = "int"
  }
}

sink {
  Mysql {
    # Replace with your MySQL connection details
    url = "jdbc:mysql://localhost:3306/test"
    driver = "com.mysql.cj.jdbc.Driver"
    username = "root"
    password = "your_password"
    # Specify the table to write to
    table = "test_table"
    # Optional: Enable exactly-once semantics if needed
    # is_exactly_once = true
    # xa_data_source_class_name = "com.mysql.cj.jdbc.MysqlXADataSource"
  }
}
```

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

### JSON Configuration Example

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

An example of a SeaTunnel configuration file written in JSON format. Ensure the file name ends with '.json' when using this 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"
    }
  ]
}

```

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

### Start Socket Server with Netcat

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

Use netcat to start a listener on the specified port to receive data. This command should be run before starting the SeaTunnel job.

```bash
nc -l 9999
```

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

### Get Overview of All Running Jobs

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

Use this GET request to get a summary of all jobs and their current statuses. The response includes job details and metrics.

```http
GET **/hazelcast/rest/maps/running-jobs**
```

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

### TDengine URL Example

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

Example of a TDengine JDBC URL.

```text
jdbc:TAOS-RS://localhost:6041/
```

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

### Sink Common Options Example

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

Demonstrates the usage of `plugin_input` and `parallelism` in a SeaTunnel pipeline configuration. This example shows how to define sources, transformations, and sinks, specifying input and output plugins for data flow control.

```hocon
source {
    FakeSourceStream {
      parallelism = 2
      plugin_output = "fake"
      field_name = "name,age"
    }
}

transform {
    Filter {
      plugin_input = "fake"
      fields = [name]
      plugin_output = "fake_name"
    }
    Filter {
      plugin_input = "fake"
      fields = [age]
      plugin_output = "fake_age"
    }
}

sink {
    Console {
      plugin_input = "fake_name"
    }
    Console {
      plugin_input = "fake_age"
    }
}
```

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

### IoTDB Source Configuration Example

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

This snippet shows a complete configuration for the IoTDB source connector within a SeaTunnel job. It includes environment settings, source configuration with connection details, SQL query, and schema definition, as well as a console sink for output.

```hocon
env {
  parallelism = 2
  job.mode = "BATCH"
}

source {
  IoTDB {
    node_urls = "localhost:6667"
    username = "root"
    password = "root"
    sql = "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"
    schema {
      fields {
        ts = timestamp
        device_name = string
        temperature = float
        moisture = bigint
        c_int = int
        c_bigint = bigint
        c_float = float
        c_double = double
        c_string = string
        c_boolean = boolean
      }
    }
  }
}

sink {
  Console {
  }
}

```