跳到主要内容
版本:Next

Sls

Sls source connector

支持的引擎

Spark
Flink
Seatunnel Zeta

主要特性

描述

从阿里云Sls日志服务中读取数据。

支持的数据源信息

为了使用Sls连接器,需要以下依赖关系。 它们可以通过install-plugin.sh或Maven中央存储库下载。

数据源支持的版本Maven
SlsUniversalDownload

Source Options

NameTypeRequiredDefaultDescription
projectStringYes-阿里云 Sls 项目
logstoreStringYes-阿里云 Sls 日志库
endpointStringYes-阿里云访问服务点
access_key_idStringYes-阿里云访问用户ID
access_key_secretStringYes-阿里云访问用户密码
start_modeStartMode[earliest],[group_cursor],[latest]Nogroup_cursor消费者的初始消费模式
consumer_groupStringNoSeaTunnel-Consumer-GroupSls消费者组id,用于区分不同的消费者组
auto_cursor_resetCursorMode[begin],[end]Noend当消费者组中没有记录读取游标时,初始化读取游标
batch_sizeIntNo1000每次从SLS中读取的数据量
partition-discovery.interval-millisLongNo-1动态发现主题和分区的间隔

任务示例

简单示例

此示例读取sls的logstore1的数据并将其打印到客户端。如果您尚未安装和部署SeaTunnel,则需要按照安装SeaTunnel中的说明安装和部署SeaTunnel。然后按照[快速启动SeaTunnel引擎](../../Start-v2/locale/Quick-Start SeaTunnel Engine.md)中的说明运行此作业。

创建RAM用户及授权, 请确认RAM用户有足够的权限来读取及管理数据,参考:RAM自定义授权示例

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