import ChangeLog from ‘../changelog/connector-file-ftp.md’;
Ftp 文件 Source 连接器
Spark
Flink
SeaTunnel Zeta
[x] 多模态
使用二进制文件格式读取和写入任何格式的文件,例如视频、图片等。简而言之,任何文件都可以同步到目标位置。
[x] 批处理
[ ] 流处理
[ ] 精确一次处理
[x] 列投影
[x] 并行度
[ ] 支持用户自定义分片
[x] 文件格式类型
从 FTP 文件服务器读取数据。
:::提示
如果您使用 Spark/Flink,为了使用此连接器,您必须确保您的 Spark/Flink 集群已经集成了 Hadoop。测试的 Hadoop 版本为 2.x。 如果您使用 SeaTunnel Engine,当您下载并安装 SeaTunnel Engine 时,它会自动集成 Hadoop 的 jar 包。您可以在 ${SEATUNNEL_HOME}/lib 目录下检查 jar 包以确认这一点。
:::
| 名称 | 类型 | 是否必填 | 默认值 |
|---|---|---|---|
| host | string | 是 | - |
| port | int | 是 | - |
| user | string | 是 | - |
| password | string | 是 | - |
| path | string | 是 | - |
| file_format_type | string | 是 | - |
| connection_mode | string | 否 | active_local |
| remote_verification_enabled | boolean | no | true |
| delimiter/field_delimiter | string | 否 | \001 |
| read_columns | list | 否 | - |
| parse_partition_from_path | boolean | 否 | true |
| date_format | string | 否 | yyyy-MM-dd |
| datetime_format | string | 否 | yyyy-MM-dd HH:mm:ss |
| time_format | string | 否 | HH:mm:ss |
| skip_header_row_number | long | 否 | 0 |
| schema | config | 否 | - |
| sheet_name | string | 否 | - |
| xml_row_tag | string | 否 | - |
| xml_use_attr_format | boolean | 否 | - |
| csv_use_header_line | boolean | 否 | false |
| file_filter_pattern | string | 否 | - |
| compress_codec | string | 否 | none |
| archive_compress_codec | string | 否 | none |
| encoding | string | 否 | UTF-8 |
| null_format | string | 否 | - |
| binary_chunk_size | int | 否 | 1024 |
| binary_complete_file_mode | boolean | 否 | false |
| discovery_mode | string | 否 | once |
| scan_interval | string | 否 | 10S |
| start_mode | string | 否 | earliest |
| sync_mode | string | 否 | full |
| target_path | string | 否 | - |
| target_hadoop_conf | map | 否 | - |
| update_strategy | string | 否 | distcp |
| compare_mode | string | 否 | len_mtime |
| common-options | 否 | - | |
| file_filter_modified_start | string | 否 | - |
| file_filter_modified_end | string | 否 | - |
| quote_char | string | 否 | " |
| escape_char | string | 否 | - |
| metalake_type | string | 否 | gravitino |
| recursive_file_scan | boolean | 否 | true |
| sort_files_by_modification_time | boolean | 否 | false |
目标 FTP 主机地址,必填项。
目标 FTP 端口,必填项。
目标 FTP 用户名,必填项。
目标 FTP 密码,必填项。
源文件路径。
是否启用FTP数据通道的远程主机验证。默认值为 true。
文件过滤模式,用于过滤文件。若只想根据文件名称筛选,则直接写文件名称的正则;若同时想根据文件目录进行过滤,则表达式以path起始。
该模式遵循标准正则表达式。详情请参考:https://en.wikipedia.org/wiki/Regular_expression. 以下是一些示例。
若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
文件类型,支持以下文件类型:
text csv parquet orc json excel xml binary markdown
如果您将文件类型指定为 json,您还需要指定 schema 选项以告诉连接器如何将数据解析为您所需的行。
例如:
上游数据如下:
{"code": 200, "data": "get success", "success": true}
您应按如下方式指定 schema:
schema { fields { code = int data = string success = boolean } }
连接器将生成如下数据:
| code | data | success |
|---|---|---|
| 200 | get success | true |
如果您将文件类型指定为 text 或 csv,您可以选择是否指定 schema 信息。
例如,上游数据如下:
tyrantlucifer#26#male
如果您不指定数据 schema,连接器将按如下方式处理上游数据:
| content |
|---|
| tyrantlucifer#26#male |
如果您指定数据 schema,您还需要指定 field_delimiter 选项(CSV 文件类型除外)。
您应按如下方式指定 schema 和分隔符:
field_delimiter = "#" schema { fields { name = string age = int gender = string } }
连接器将生成如下数据:
| name | age | gender |
|---|---|---|
| tyrantlucifer | 26 | male |
如果您将文件类型指定为 binary,SeaTunnel 可以同步任何格式的文件, 例如压缩包、图片等。简而言之,任何文件都可以同步到目标位置。 在这种情况下,您需要确保源和接收端同时使用 binary 格式进行文件同步。 您可以在下面的示例中找到具体用法。
如果您将文件类型指定为 markdown,SeaTunnel 可以解析 markdown 文件并提取结构化数据。 markdown 解析器提取各种元素,包括标题、段落、列表、代码块、表格等。 每个元素都转换为具有以下架构的行:
element_id:元素的唯一标识符element_type:元素类型(Heading、Paragraph、ListItem 等)heading_level:标题级别(1-6,非标题元素为 null)text:元素的文本内容page_number:页码(默认:1)position_index:文档中的位置索引parent_id:父元素的 IDchild_ids:子元素 ID 的逗号分隔列表当 markdown_rag_metadata_enabled 设置为 true 时,SeaTunnel 会在 child_ids 之后追加以下 RAG 元数据字段:
source_uri:源文件路径或 URIdocument_id:由 source_uri 派生的稳定文档标识符chunk_id:由文档标识、chunk 顺序和内容哈希派生的稳定 chunk 标识符chunk_index:解析后文档中的一基 chunk 顺序content_hash:已输出 text 值的 SHA-256 哈希该选项默认值为 false,因此只有显式启用后才会改变原始 Markdown schema。
注意:Markdown 格式仅支持读取,不支持写入。
目标 FTP 连接模式,默认为主动模式,支持以下模式:
active_local passive_local
FTP 控制连接的字符编码。默认为 UTF-8。
当文件路径包含特殊字符(如 $、空格、中文字符等)时,需要设置为 UTF-8 以确保路径能够正确解析。
例如:/data/whale_ops/share/$Fund-Product/DA - SANY (三一)/Daily/2025.08.18/file.xlsx
delimiter 参数将在 2.3.5 版本后弃用,请使用 field_delimiter 代替。
仅在文件格式为 text 时需要配置。
字段分隔符,用于告诉连接器如何切分字段。
默认值为 \001,与 Hive 的默认分隔符相同。
控制是否从文件路径中解析分区键和值。
例如,如果您从路径 ftp://hadoop-cluster/tmp/seatunnel/parquet/name=tyrantlucifer/age=26 读取文件,
文件中的每条记录数据将添加以下两个字段:
| name | age |
|---|---|
| tyrantlucifer | 26 |
提示:不要在 schema 选项中定义分区字段
日期类型格式,用于告诉连接器如何将字符串转换为日期,支持以下格式:
yyyy-MM-dd yyyy.MM.dd yyyy/MM/dd
默认值为 yyyy-MM-dd
日期时间类型格式,用于告诉连接器如何将字符串转换为日期时间,支持以下格式:
yyyy-MM-dd HH:mm:ss yyyy.MM.dd HH:mm:ss yyyy/MM/dd HH:mm:ss yyyyMMddHHmmss
默认值为 yyyy-MM-dd HH:mm:ss
时间类型格式,用于告诉连接器如何将字符串转换为时间,支持以下格式:
HH:mm:ss HH:mm:ss.SSS
默认值为 HH:mm:ss
跳过前几行,仅适用于 txt 和 csv 文件。
例如,设置如下:
skip_header_row_number = 2
SeaTunnel 将从源文件中跳过前 2 行。
仅在文件格式类型为 text、json、excel、xml 或 csv(或其他无法从元数据中读取 schema 的格式)时需要配置。
上游数据的 schema 信息。更多详情请参考 Schema 特性。
元数据服务中的表标识符,用于获取表结构。对于 Gravitino,格式应为 {catalog}.{database}.{table},例如 mysql-catalog.test_db.users。
当指定此参数时,连接器将从外部元数据服务获取表结构,而不是使用手动定义的 columns。
当使用 Gravitino 作为元数据源时,Gravitino 的列类型会自动转换为 SeaTunnel 数据类型。详细的类型映射信息请参考 Gravitino 类型映射。
更多信息请参考 元数据 SPI。
数据源的读取列列表,用户可以使用它来实现字段投影。
读取工作簿中的工作表,仅在文件格式类型为 excel 时使用。
仅在文件格式为 xml 时需要配置。
指定 XML 文件中数据行的标签名称。
仅在文件格式为 xml 时需要配置。
指定是否使用标签属性格式处理数据。
仅在文件格式为 csv 时可以选择配置。 是否使用标题行来解析文件, 标题行 与 RFC 4180 匹配
文件的压缩编解码器,支持的详细信息如下:
lzo nonelzo nonelzo none归档文件的压缩编解码器,支持的详细信息如下:
| archive_compress_codec | 文件格式 | 归档压缩后缀 |
|---|---|---|
| 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
仅在文件格式类型为 json、text、csv、xml 时使用。 读取文件的编码。此参数将通过 Charset.forName(encoding) 解析。
仅在文件格式类型为 text 时使用。 用于定义哪些字符串可以表示为 null。
例如:\N
仅在 file_format_type 为 binary 时使用。
读取二进制文件的块大小(以字节为单位)。默认为 1024 字节。较大的值可能会提高大文件的性能,但会使用更多内存。
仅在 file_format_type 为 binary 时使用。
是否将完整文件作为单个块读取,而不是分割成块。启用时,整个文件内容将一次性读入内存。默认为 false。
文件发现模式,支持:once(默认)、continuous。
once:启动时枚举一次文件并结束(有界)。continuous:作业保持运行,周期性扫描路径并在运行时处理新增/变更文件(无界)。当前实现中,discovery_mode=continuous 需要配合 sync_mode=update(仅 binary)使用,以避免重复传输。
仅在 discovery_mode=continuous 时使用。周期性扫描间隔,取值必须大于 0。推荐使用简写格式 10S、30S(大小写不敏感,例如 10s);同时兼容 ISO-8601 格式 PT10S、PT30S。默认 10S。
仅在 discovery_mode=continuous 时使用,支持:earliest(默认)、latest。
earliest:启动时读取已有文件。latest:仅处理作业启动后修改的新文件。文件同步模式,支持:full(默认)、update。 当 update 时,对源/目标进行对比,只读取新增/变更文件(目前仅支持 file_format_type=binary)。
性能注意事项
getFileStatus 用于对比。要求 / 限制
target_path 通常应与 sink 的 path 一致(同一文件系统且相对路径结构一致)。update_strategy=distcp 时,依赖源/目标端时钟同步,否则可能误判。compare_mode=checksum 时,需要文件系统支持 checksum;若无法获取 checksum,SeaTunnel 会降级为内容比较(开销更大)并打印告警日志。示例:
sync_mode = "update" file_format_type = "binary" target_path = "/path/to/your/sink/path" update_strategy = "distcp" compare_mode = "len_mtime"
仅在 sync_mode=update 时使用。目标端基础路径(通常应与 sink 的 path 一致),用于对比同相对路径文件。
仅在 sync_mode=update 时使用。目标端 Hadoop 配置(可选),可在其中设置 fs.defaultFS 覆盖目标 defaultFS。
仅在 sync_mode=update 时使用。支持:distcp(默认)、strict。
仅在 sync_mode=update 时使用。支持:len_mtime(默认)、checksum(仅在 update_strategy=strict 时可用)。
按照最后修改时间过滤文件。 要过滤的开始时间(包括改时间),时间格式是:yyyy-MM-dd HH:mm:ss。
按照最后修改时间过滤文件。 要过滤的结束时间(不包括改时间),时间格式是:yyyy-MM-dd HH:mm:ss。
用于包裹 CSV 字段的单字符,可保证包含逗号、换行符或引号的字段被正确解析。
用于在 CSV 字段内转义引号或其他特殊字符,使其不会结束字段。
是否递归扫描子目录。 如果设置为 false,将忽略子目录,仅扫描指定路径下的文件。
是否按修改时间降序排序文件。默认值为 false。 启用后,文件将按修改时间排序(最新的在前)。适用于以下场景:
源插件的通用参数,详情请参考 源通用选项。
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 = "#" }
FtpFile { tables_configs = [ { schema { table = "student" } path = "/tmp/seatunnel/sink/text" host = "192.168.31.48" port = 21 user = tyrantlucifer password = tianchao file_format_type = "parquet" }, { schema { table = "teacher" } path = "/tmp/seatunnel/sink/text" host = "192.168.31.48" port = 21 user = tyrantlucifer password = tianchao file_format_type = "parquet" } ] }
FtpFile { tables_configs = [ { schema { fields { name = string age = int } } path = "/apps/hive/demo/student" file_format_type = "json" }, { schema { fields { name = string age = int } } path = "/apps/hive/demo/teacher" file_format_type = "json" } }
env { parallelism = 1 job.mode = "BATCH" } source { FtpFile { host = "192.168.31.48" port = 21 user = tyrantlucifer password = tianchao path = "/seatunnel/read/binary/" file_format_type = "binary" binary_chunk_size = 2048 binary_complete_file_mode = false } } sink { // 您可以将本地文件传输到 s3/hdfs/oss 等。 FtpFile { host = "192.168.31.48" port = 21 user = tyrantlucifer password = tianchao path = "/seatunnel/read/binary2/" file_format_type = "binary" } }
sync_mode=update 会对比 source 与 target_path,仅读取新增/变更文件。 多数情况下,target_path 需要与 sink 的 path 对齐(同一文件系统、相同相对路径)。
env { parallelism = 1 job.mode = "BATCH" } source { FtpFile { host = "192.168.31.48" port = 21 user = tyrantlucifer password = tianchao path = "/seatunnel/read/binary/" file_format_type = "binary" sync_mode = "update" target_path = "/seatunnel/read/binary2/" update_strategy = "distcp" compare_mode = "len_mtime" } } sink { FtpFile { host = "192.168.31.48" port = 21 user = tyrantlucifer password = tianchao path = "/seatunnel/read/binary2/" tmp_path = "/seatunnel/read/binary2-tmp/" file_format_type = "binary" } }
discovery_mode=continuous 会让作业保持运行,并按间隔持续扫描路径发现新/变更文件(长跑作业,推荐使用 job.mode="STREAMING")。
注意: discovery_mode=continuous 当前需要配合 sync_mode="update"(仅支持 binary)使用,以避免重复传输而不引入无限增长的“已处理状态”。同时 target_path 通常应与 sink 的 path 保持一致(同一文件系统、相同相对路径)。
env { parallelism = 1 job.mode = "STREAMING" } source { FtpFile { host = "192.168.31.48" port = 21 user = tyrantlucifer password = tianchao path = "/seatunnel/watch/src/" file_format_type = "binary" discovery_mode = "continuous" scan_interval = "10S" start_mode = "latest" sync_mode = "update" target_path = "/seatunnel/watch/dst/" update_strategy = "distcp" compare_mode = "len_mtime" } } sink { FtpFile { host = "192.168.31.48" port = 21 user = tyrantlucifer password = tianchao path = "/seatunnel/watch/dst/" tmp_path = "/seatunnel/watch/dst-tmp/" file_format_type = "binary" } }
env { parallelism = 1 job.mode = "BATCH" } source { FtpFile { host = "192.168.31.48" port = 21 user = tyrantlucifer password = tianchao path = "/seatunnel/read/binary/" file_format_type = "binary" // 文件示例 abcD2024.csv file_filter_pattern = "abc[DX]*.*" } } sink { Console { } }