Sls

Sls source connector

Support Those Engines

Spark
Flink
SeaTunnel Zeta

Key Features

• batch
• stream
• exactly-once
• column projection
• parallelism
• support user-defined split

Description

The Sls source connector reads logs from Alibaba Cloud Simple Log Service (SLS). It can run in batch or streaming jobs and reads SLS shards in parallel. In streaming mode, SeaTunnel stores the SLS cursor during checkpoint completion, so a restarted job can continue from the committed cursor.

You can read logs in two ways:

• Configure schema to parse named SLS log fields into SeaTunnel columns.
• Omit schema to read each SLS log as one JSON string in a single content column.

Supported DataSource Info

To use the Sls connector, download the following dependency by using install-plugin.sh or from the Maven central repository.

Datasource	Supported Versions	Maven
Sls	Universal	Download

Source Options

Name	Type	Required	Default	Description
endpoint	String	Yes	-	Alibaba Cloud SLS endpoint, for example cn-hangzhou.log.aliyuncs.com or an intranet endpoint.
project	String	Yes	-	Alibaba Cloud SLS project.
logstore	String	Yes	-	Alibaba Cloud SLS logstore.
access_key_id	String	Yes	-	Alibaba Cloud AccessKey ID.
access_key_secret	String	Yes	-	Alibaba Cloud AccessKey secret.
start_mode	earliest, group_cursor, latest	No	group_cursor	Initial cursor mode. earliest starts from the beginning, latest starts from the end, and group_cursor uses the consumer group's checkpoint.
consumer_group	String	No	SeaTunnel-Consumer-Group	SLS consumer group name. Use different values when separate jobs must keep independent cursors.
auto_cursor_reset	begin, end	No	end	Cursor position used when start_mode = group_cursor but the consumer group has no checkpoint yet.
batch_size	Int	No	1000	Maximum logs pulled from each shard in one request.
partition-discovery.interval-millis	Long	No	-1	Interval for discovering SLS shard changes. A value less than or equal to 0 disables periodic discovery.
schema	Config	No	-	SeaTunnel schema for parsing SLS log fields. If omitted, the connector outputs one content string column containing the full log as JSON.

Notes

• The configured RAM user must have permission to read the target project, logstore, shards, consumer groups, checkpoints, and logs.
• In batch mode, the connector reads the currently assigned shards and then finishes. Use streaming mode for continuous log consumption.
• Do not print access_key_secret in logs or job descriptions.

Task Examples

Read Logs with an Explicit Schema

env {
  parallelism = 1
  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"
    consumer_group = "seatunnel-sls-demo"
    start_mode = "group_cursor"
    auto_cursor_reset = "begin"
    batch_size = 1000
    schema = {
      fields = {
        id = "int"
        name = "string"
        description = "string"
        weight = "string"
      }
    }
  }
}

sink {
  Console {}
}

Read Logs Without a Schema

When schema is not configured, each output row has one content field. The field value is a JSON string built from the SLS log contents.

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

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

sink {
  Console {}
}

Changelog

Change Log

Change	Commit	Version
[Fix][Connector-V2] Return catalog tables from Aerospike and SLS sinks (#12234)	https://github.com/apache/seatunnel/commit/ea1fb0d85	3.0.0
[Chore] fix typos filed -> field (#9757)	https://github.com/apache/seatunnel/commit/e3e1c67d29	2.3.12
[Improve][API] Optimize the enumerator API semantics and reduce lock calls at the connector level (#9671)	https://github.com/apache/seatunnel/commit/9212a77140	2.3.12
[improve] sls options (#9260)	https://github.com/apache/seatunnel/commit/126164508b	2.3.11
[Feature][Checkpoint] Add check script for source/sink state class serialVersionUID missing (#9118)	https://github.com/apache/seatunnel/commit/4f5adeb1c7	2.3.11
[Improve] restruct connector common options (#8634)	https://github.com/apache/seatunnel/commit/f3499a6eeb	2.3.10
[Feature]Check Chinese comments in the code (#8319)	https://github.com/apache/seatunnel/commit/d58fce1caf	2.3.9
[Improve][dist]add shade check rule (#8136)	https://github.com/apache/seatunnel/commit/51ef800016	2.3.9
[Improve][Sls] Add sls sink connector、e2e、doc (#7830)	https://github.com/apache/seatunnel/commit/048c47d966	2.3.9
[Fix][Connector-V2] Fix some throwable error not be caught (#7657)	https://github.com/apache/seatunnel/commit/e19d73282e	2.3.8
[Feature][Connector-V2] add Aliyun SLS connector #3733 (#7348)	https://github.com/apache/seatunnel/commit/527c7c7b5f	2.3.7