import ChangeLog from ‘../changelog/connector-file-s3.md’;

S3File

S3文件数据源连接器

支持的引擎

Spark
Flink
SeaTunnel Zeta

主要特性

[x] 多模态
使用二进制文件格式读取和写入任何格式的文件，例如视频、图片等。简而言之，任何文件都可以同步到目标位置。
[x] 批处理
[ ] 流处理
[x] 精确一次
在一次pollNext调用中读取分片中的所有数据。将读取的分片保存在快照中。
[x] 列投影
[x] 并行度
[ ] 支持用户定义的分片
[x] 文件格式类型
- [x] text
- [x] csv
- [x] parquet
- [x] orc
- [x] json
- [x] excel
- [x] xml
- [x] binary
- [x] markdown

描述

从aws s3文件系统读取数据。

支持的数据源信息

数据源	支持的版本
S3	current

依赖

如果您使用spark/flink，为了使用此连接器，您必须确保您的spark/flink集群已经集成了hadoop。测试过的hadoop版本是2.x。
如果您使用SeaTunnel Zeta，它在您下载和安装SeaTunnel Zeta时会自动集成hadoop jar。您可以检查${SEATUNNEL_HOME}/lib下的jar包来确认这一点。
要使用此连接器，您需要将hadoop-aws-3.1.4.jar和aws-java-sdk-bundle-1.12.692.jar放在${SEATUNNEL_HOME}/lib目录中。

数据类型映射

数据类型映射与正在读取的文件类型相关，我们支持以下文件类型：

text csv parquet orc json excel xml

JSON文件类型

如果您将文件类型指定为json，您还应该指定schema选项来告诉连接器如何将数据解析为您想要的行。

例如：

上游数据如下：


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

您也可以在一个文件中保存多条数据，并用换行符分隔：


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

您应该按如下方式指定schema：


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

连接器将生成如下数据：

code	data	success
200	get success	true

文本或CSV文件类型

如果您将file_format_type设置为text、excel、csv、xml。那么需要设置schema字段来告诉连接器如何将数据解析为行。

如果您设置了schema字段，您还应该设置选项field_delimiter，除非file_format_type是csv、xml、excel

您可以按如下方式设置schema和分隔符：


field_delimiter = "#"
schema {
    fields {
        name = string
        age = int
        gender = string 
    }
}

连接器将生成如下数据：

name	age	gender
tyrantlucifer	26	male

Orc文件类型

如果您将文件类型指定为parquet orc，则不需要schema选项，连接器可以自动找到上游数据的schema。

Orc数据类型	SeaTunnel数据类型
BOOLEAN	BOOLEAN
INT	INT
BYTE	BYTE
SHORT	SHORT
LONG	LONG
FLOAT	FLOAT
DOUBLE	DOUBLE
BINARY	BINARY
STRING VARCHAR CHAR	STRING
DATE	LOCAL_DATE_TYPE
TIMESTAMP	LOCAL_DATE_TIME_TYPE
DECIMAL	DECIMAL
LIST(STRING)	STRING_ARRAY_TYPE
LIST(BOOLEAN)	BOOLEAN_ARRAY_TYPE
LIST(TINYINT)	BYTE_ARRAY_TYPE
LIST(SMALLINT)	SHORT_ARRAY_TYPE
LIST(INT)	INT_ARRAY_TYPE
LIST(BIGINT)	LONG_ARRAY_TYPE
LIST(FLOAT)	FLOAT_ARRAY_TYPE
LIST(DOUBLE)	DOUBLE_ARRAY_TYPE
Map<K,V>	MapType，K和V的类型将转换为SeaTunnel类型
STRUCT	SeaTunnelRowType

Parquet文件类型

如果您将文件类型指定为parquet orc，则不需要schema选项，连接器可以自动找到上游数据的schema。

Parquet数据类型	SeaTunnel数据类型
INT_8	BYTE
INT_16	SHORT
DATE	DATE
TIMESTAMP_MILLIS	TIMESTAMP
INT64	LONG
INT96	TIMESTAMP
BINARY	BYTES
FLOAT	FLOAT
DOUBLE	DOUBLE
BOOLEAN	BOOLEAN
FIXED_LEN_BYTE_ARRAY	TIMESTAMP DECIMAL
DECIMAL	DECIMAL
LIST(STRING)	STRING_ARRAY_TYPE
LIST(BOOLEAN)	BOOLEAN_ARRAY_TYPE
LIST(TINYINT)	BYTE_ARRAY_TYPE
LIST(SMALLINT)	SHORT_ARRAY_TYPE
LIST(INT)	INT_ARRAY_TYPE
LIST(BIGINT)	LONG_ARRAY_TYPE
LIST(FLOAT)	FLOAT_ARRAY_TYPE
LIST(DOUBLE)	DOUBLE_ARRAY_TYPE
Map<K,V>	MapType，K和V的类型将转换为SeaTunnel类型
STRUCT	SeaTunnelRowType

选项

名称	类型	是否必需	默认值	描述
path	string	是	-	需要读取的s3路径，可以有子路径，但子路径需要满足一定的格式要求。具体要求可以参考“parse_partition_from_path”选项
file_format_type	string	是	-	文件类型，支持以下文件类型：`text` `csv` `parquet` `orc` `json` `excel` `xml` `binary` `markdown`
bucket	string	是	-	s3文件系统的bucket地址，例如：`s3n://seatunnel-test`，如果您使用`s3a`协议，此参数应为`s3a://seatunnel-test`。
fs.s3a.endpoint	string	是	-	fs s3a端点
fs.s3a.aws.credentials.provider	string	是	com.amazonaws.auth.InstanceProfileCredentialsProvider	s3a的认证方式。我们目前只支持`org.apache.hadoop.fs.s3a.SimpleAWSCredentialsProvider`和`com.amazonaws.auth.InstanceProfileCredentialsProvider`。有关凭据提供程序的更多信息，您可以查看Hadoop AWS文档
read_columns	list	否	-	数据源的读取列列表，用户可以使用它来实现字段投影。支持列投影的文件类型如下所示：`text` `csv` `parquet` `orc` `json` `excel` `xml`。如果用户想在读取`text` `json` `csv`文件时使用此功能，必须配置“schema”选项。
access_key	string	否	-	仅在`fs.s3a.aws.credentials.provider = org.apache.hadoop.fs.s3a.SimpleAWSCredentialsProvider`时使用
secret_key	string	否	-	仅在`fs.s3a.aws.credentials.provider = org.apache.hadoop.fs.s3a.SimpleAWSCredentialsProvider`时使用
hadoop_s3_properties	map	否	-	如果您需要添加其他选项，可以在此处添加并参考此链接
delimiter/field_delimiter	string	否	\001	字段分隔符，用于告诉连接器在读取文本文件时如何切分字段。默认`\001`，与hive的默认分隔符相同。
row_delimiter	string	否	\n	行分隔符，用于告诉连接器在读取文本文件时如何切分行。默认`\n`。
parse_partition_from_path	boolean	否	true	控制是否从文件路径解析分区键和值。例如，如果您从路径`s3n://hadoop-cluster/tmp/seatunnel/parquet/name=tyrantlucifer/age=26`读取文件。文件中的每条记录数据都将添加这两个字段：name=“tyrantlucifer”，age=16
date_format	string	否	yyyy-MM-dd	日期类型格式，用于告诉连接器如何将字符串转换为日期，支持以下格式：`yyyy-MM-dd` `yyyy.MM.dd` `yyyy/MM/dd`。默认`yyyy-MM-dd`
datetime_format	string	否	yyyy-MM-dd HH:mm:ss	日期时间类型格式，用于告诉连接器如何将字符串转换为日期时间，支持以下格式：`yyyy-MM-dd HH:mm:ss` `yyyy.MM.dd HH:mm:ss` `yyyy/MM/dd HH:mm:ss` `yyyyMMddHHmmss`
time_format	string	否	HH:mm:ss	时间类型格式，用于告诉连接器如何将字符串转换为时间，支持以下格式：`HH:mm:ss` `HH:mm:ss.SSS`
skip_header_row_number	long	否	0	跳过前几行，但仅适用于txt和csv。例如，设置如下：`skip_header_row_number = 2`。然后SeaTunnel将跳过源文件的前2行
csv_use_header_line	boolean	否	false	是否使用标题行来解析文件，仅在file_format为`csv`且文件包含符合RFC 4180的标题行时使用
schema	config	否	-	上游数据的schema。更多详情请参考 Schema 特性。
sheet_name	string	否	-	读取工作簿的工作表，仅在file_format为excel时使用。
xml_row_tag	string	否	-	指定XML文件中数据行的标签名称，仅对XML文件有效。
xml_use_attr_format	boolean	否	-	指定是否使用标签属性格式处理数据，仅对XML文件有效。
compress_codec	string	否	none
archive_compress_codec	string	否	none
enable_file_split	boolean	否	false	开启大文件拆分以提升并行度。仅支持 `text`/`csv`/`json`/`parquet` 且非压缩格式（`compress_codec=none` 且 `archive_compress_codec=none`）。
file_split_size	long	否	134217728	`enable_file_split=true` 时生效，单位字节。`text`/`csv`/`json` 按 `file_split_size` 拆分并对齐到下一个 `row_delimiter`；`parquet` 以 RowGroup 为拆分单位，不会切开 RowGroup。
encoding	string	否	UTF-8
null_format	string	否	-	仅在file_format_type为text时使用。null_format用于定义哪些字符串可以表示为null。例如：`\N`
binary_chunk_size	int	否	1024	仅在file_format_type为binary时使用。读取二进制文件的块大小（以字节为单位）。默认为1024字节。较大的值可能会提高大文件的性能，但会使用更多内存。
binary_complete_file_mode	boolean	否	false	仅在file_format_type为binary时使用。是否将完整文件作为单个块读取，而不是分割成块。启用时，整个文件内容将一次性读入内存。默认为false。
file_filter_pattern	string	否		过滤模式，用于过滤文件。
filename_extension	string	否	-	过滤文件名扩展名，用于过滤具有特定扩展名的文件。例如：`csv` `.txt` `json` `.xml`。
common-options		否	-	数据源插件通用参数，请参考数据源通用选项了解详情。
quote_char	string	否	"	用于包裹 CSV 字段的单字符，可保证包含逗号、换行符或引号的字段被正确解析。
escape_char	string	否	-	用于在 CSV 字段内转义引号或其他特殊字符，使其不会结束字段。
metalake_type	string	否	gravitino	Metalake 服务类型，目前支持 `gravitino`。
recursive_file_scan	boolean	否	true	是否递归扫描子目录。如果设置为 `false`，将忽略子目录，仅扫描指定路径下的文件。
sort_files_by_modification_time	boolean	否	false	是否按修改时间降序排序文件。启用此选项后，在读取不断演化的 schema 时可确保 schema 推断使用最新的文件。

delimiter/field_delimiter [string]

delimiter参数将在2.3.5版本后弃用，请使用field_delimiter代替。

row_delimiter [string]

仅在 file_format 为 text 时需要配置。

行分隔符，用于告诉连接器如何分割行。

默认 \n。

quote_char [string]

用于包裹 CSV 字段的单字符，可保证包含逗号、换行符或引号的字段被正确解析。

escape_char [string]

用于在 CSV 字段内转义引号或其他特殊字符，使其不会结束字段。

recursive_file_scan [boolean]

是否递归扫描子目录。如果设置为 false，将忽略子目录，仅扫描指定路径下的文件。

sort_files_by_modification_time [boolean]

是否按修改时间降序排序文件。默认值为 false。启用后，文件将按修改时间排序（最新的在前）。适用于以下场景：

读取具有不断演化的 schema 的文件，且希望 schema 推断使用最新的文件
需要按时间顺序处理文件

file_filter_pattern [string]

文件过滤模式，用于过滤文件。若只想根据文件名称筛选，则直接写文件名称的正则；若同时想根据文件目录进行过滤，则表达式以path起始。

该模式遵循标准正则表达式。详情请参考正则表达式。以下是一些示例。

若path为/data/seatunnel,且文件结构示例：

/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

匹配规则示例：

示例1：匹配所有.txt文件，正则表达式：

.*.txt

此示例匹配的结果是：

/data/seatunnel/20241001/report.txt

示例2：匹配所有以abc开头的文件，正则表达式：

abc.*

此示例匹配的结果是：

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

示例3：匹配20241007文件夹下所有以 abc 开头的文件，且第四个字符为 h 或 g，正则表达式：

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

此示例匹配的结果是：

/data/seatunnel/20241007/abch202410.csv

示例4：匹配以202410开头的第三级文件夹和以.csv结尾的文件，正则表达式：

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

此示例匹配的结果是：

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

enable_file_split [boolean]

开启大文件拆分功能，默认 false。仅支持 csv/text/json/parquet 且非压缩格式（compress_codec=none 且 archive_compress_codec=none）。

text/csv/json：按 file_split_size 拆分并对齐到下一个 row_delimiter，避免切开一行/一条记录。
parquet：以 RowGroup 为逻辑拆分单位，不会切开 RowGroup。

使用建议

适合：读取少量大文件，并希望通过更高并行度提升吞吐。
不建议：读取大量小文件，或并行度较低的场景（拆分会带来额外的枚举/调度开销）。

限制说明

不支持压缩文件（compress_codec != none）或归档文件（archive_compress_codec != none），会自动回退为不拆分，并打印 WARN 日志提示。
对于 text/csv/json，实际 split 的大小可能略大于 file_split_size（因为需要对齐到下一个 row_delimiter）。
对于 json，仅支持 JSON Lines（每行一个 JSON 对象）的切分读取。
启用切分后，数据全局顺序不保证（split 可能并行处理导致输出顺序交错）。如需严格有序，请设置 parallelism=1 或关闭切分。

file_split_size [long]

enable_file_split=true 时生效，单位字节。默认 128MB（134217728）。

调优建议

建议从默认值（128MB）开始：如果并行度未充分利用可适当调小；如果 split 数量过多可适当调大。
经验公式：file_split_size ≈ file_size / 期望并行度。

compress_codec [string]

文件的压缩编解码器，支持的详细信息如下所示：

txt: lzo none
json: lzo none
csv: lzo none
orc/parquet: 自动识别压缩类型，无需额外设置。

archive_compress_codec [string]

归档文件的压缩编解码器，支持的详细信息如下所示：

archive_compress_codec	file_format	archive_compress_suffix
ZIP	txt,json,excel,xml	.zip
TAR	txt,json,excel,xml	.tar
TAR_GZ	txt,json,excel,xml	.tar.gz
GZ	txt,json,excel,xml	.gz
NONE	all	.*

注意：gz压缩的excel文件需要压缩原始文件或指定文件后缀，例如e2e.xls ->e2e_test.xls.gz

encoding [string]

仅在file_format_type为json、text、csv、xml时使用。要读取的文件的编码。此参数将由Charset.forName(encoding)解析。

binary_chunk_size [int]

仅在file_format_type为binary时使用。

读取二进制文件的块大小（以字节为单位）。默认为1024字节。较大的值可能会提高大文件的性能，但会使用更多内存。

binary_complete_file_mode [boolean]

仅在file_format_type为binary时使用。

是否将完整文件作为单个块读取，而不是分割成块。启用时，整个文件内容将一次性读入内存。默认为false。

file_format_type [string]

文件类型，支持以下文件类型：

text csv parquet orc json excel xml binary markdown

如果您将文件类型指定为 markdown，SeaTunnel 可以解析 markdown 文件并提取结构化数据。 markdown 解析器提取各种元素，包括标题、段落、列表、代码块、表格等。每个元素都转换为具有以下架构的行：

element_id：元素的唯一标识符
element_type：元素类型（Heading、Paragraph、ListItem 等）
heading_level：标题级别（1-6，非标题元素为 null）
text：元素的文本内容
page_number：页码（默认：1）
position_index：文档中的位置索引
parent_id：父元素的 ID
child_ids：子元素 ID 的逗号分隔列表

当 markdown_rag_metadata_enabled 设置为 true 时，SeaTunnel 会在 child_ids 之后追加以下 RAG 元数据字段：

source_uri：源文件路径或 URI
document_id：由 source_uri 派生的稳定文档标识符
chunk_id：由文档标识、chunk 顺序和内容哈希派生的稳定 chunk 标识符
chunk_index：解析后文档中的一基 chunk 顺序
content_hash：已输出 text 值的 SHA-256 哈希

该选项默认值为 false，因此只有显式启用后才会改变原始 Markdown schema。

注意：Markdown 格式仅支持读取，不支持写入。

schema [config]

仅在文件格式类型为 text、json、excel、xml 或 csv（或其他无法从元数据中读取 schema 的格式）时需要配置。

上游数据的 schema 信息。更多详情请参考 Schema 特性。

metadata_table_id [string]

元数据服务中的表标识符，用于获取表结构。对于 Gravitino，格式应为 {catalog}.{database}.{table}，例如 mysql-catalog.test_db.users。

当指定此参数时，连接器将从外部元数据服务获取表结构，而不是使用手动定义的 columns。

当使用 Gravitino 作为元数据源时，Gravitino 的列类型会自动转换为 SeaTunnel 数据类型。详细的类型映射信息请参考 Gravitino 类型映射。

更多信息请参考元数据 SPI。

示例

在此示例中，我们从s3路径s3a://seatunnel-test/seatunnel/text读取数据，此路径中的文件类型是orc。我们使用org.apache.hadoop.fs.s3a.SimpleAWSCredentialsProvider进行身份验证，因此需要access_key和secret_key。文件中的所有列都将被读取并发送到接收器。

# 定义运行时环境
env {
  parallelism = 1
  job.mode = "BATCH"
}

source {
  S3File {
    path = "/seatunnel/text"
    fs.s3a.endpoint="s3.cn-north-1.amazonaws.com.cn"
    fs.s3a.aws.credentials.provider = "org.apache.hadoop.fs.s3a.SimpleAWSCredentialsProvider"
    access_key = "xxxxxxxxxxxxxxxxx"
    secret_key = "xxxxxxxxxxxxxxxxx"
    bucket = "s3a://seatunnel-test"
    file_format_type = "orc"
  }
}

transform {
  # 如果您想获取有关如何配置seatunnel和查看转换插件完整列表的更多信息，
    # 请访问 https://seatunnel.apache.org/docs/transforms
}

sink {
  Console {}
}

使用InstanceProfileCredentialsProvider进行身份验证 S3中的文件类型是json，因此需要配置schema选项。


  S3File {
    path = "/seatunnel/json"
    bucket = "s3a://seatunnel-test"
    fs.s3a.endpoint="s3.cn-north-1.amazonaws.com.cn"
    fs.s3a.aws.credentials.provider="com.amazonaws.auth.InstanceProfileCredentialsProvider"
    file_format_type = "json"
    schema {
      fields {
        id = int
        name = string
      }
    }
  }

使用InstanceProfileCredentialsProvider进行身份验证 S3中的文件类型是json，有五个字段（id、name、age、sex、type），因此需要配置schema选项。在此作业中，我们只需要将id和name列发送到mysql。

# 定义运行时环境
env {
  parallelism = 1
  job.mode = "BATCH"
}

source {
  S3File {
    path = "/seatunnel/json"
    bucket = "s3a://seatunnel-test"
    fs.s3a.endpoint="s3.cn-north-1.amazonaws.com.cn"
    fs.s3a.aws.credentials.provider="com.amazonaws.auth.InstanceProfileCredentialsProvider"
    file_format_type = "json"
    read_columns = ["id", "name"]
    schema {
      fields {
        id = int
        name = string
        age = int
        sex = int
        type = string
      }
    }
  }
}

transform {
  # 如果您想获取有关如何配置seatunnel和查看转换插件完整列表的更多信息，
    # 请访问 https://seatunnel.apache.org/docs/transforms
}

sink {
  Console {}
}

过滤文件

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

source {
  S3File {
    path = "/seatunnel/json"
    bucket = "s3a://seatunnel-test"
    fs.s3a.endpoint="s3.cn-north-1.amazonaws.com.cn"
    fs.s3a.aws.credentials.provider="com.amazonaws.auth.InstanceProfileCredentialsProvider"
    file_format_type = "json"
    read_columns = ["id", "name"]
    // 文件示例 abcD2024.csv
    file_filter_pattern = "abc[DX]*.*"
  }
}

sink {
  Console {
  }
}

S3File

支持的引擎

主要特性

描述

支持的数据源信息

依赖

数据类型映射

JSON文件类型

文本或CSV文件类型

Orc文件类型

Parquet文件类型

选项

delimiter/field_delimiter [string]

row_delimiter [string]

quote_char [string]

escape_char [string]

recursive_file_scan [boolean]

sort_files_by_modification_time [boolean]

file_filter_pattern [string]

enable_file_split [boolean]

file_split_size [long]

compress_codec [string]

archive_compress_codec [string]

encoding [string]

binary_chunk_size [int]

binary_complete_file_mode [boolean]

file_format_type [string]

schema [config]

metadata_table_id [string]

示例

过滤文件

变更日志