據(jù)寫入 FTP 服務(wù)器的配置、參數(shù)與實現(xiàn)原理)
數(shù)據(jù)工程大數(shù)據(jù)批處理流處理【免費下載鏈接】seatunnelSeaTunnel is a next-generation super high-performance, distributed, massive data integration tool.項目地址https://gitcode.com/gh_mirrors/sea/seatunnel點擊查看免費下載FtpFile Sink 是 Apache SeaTunnel 內(nèi)置的文件型輸出連接器負(fù)責(zé)把上游 Source 或 Transform 產(chǎn)出的數(shù)據(jù)以 text、csv、parquet、orc、json、excel、xml、binary 等格式寫入 FTP 服務(wù)器指定目錄。本文以官方文檔 docs/en/connector-v2/sink/FtpFile.md 為主體結(jié)合倉庫中connector-file-ftp與connector-file-base模塊源碼完整講解該連接器的全部配置參數(shù)、運行環(huán)境要求、分區(qū)與事務(wù)機(jī)制、完整配置示例以及 2PC 提交與 FTP 文件系統(tǒng)適配的底層實現(xiàn)。讀完本文你將能夠獨立完成一個從任意 Source 到 FTP 的 SeaTunnel 作業(yè)的配置、調(diào)優(yōu)與排障。連接器簡介與適用場景FtpFile 連接器的職責(zé)非常聚焦Output data to Ftp——將 SeaTunnel 管道中的SeaTunnelRow數(shù)據(jù)流式寫入 FTP 服務(wù)器。它與同模塊的 FtpFileSource 形成一對讀寫組合適用于數(shù)據(jù)落地 FTP、與外部系統(tǒng)通過 FTP 交換文件的場景。從源碼看連接器插件名來自 FileSystemType.FTPSink 側(cè)實現(xiàn)為 FtpFileSink它繼承自文件連接器公共基類BaseFileSink因此所有文件型 Sink 的通用能力分區(qū)、事務(wù)、格式、壓縮它都天然具備。運行環(huán)境前提官方文檔對運行環(huán)境有明確提示務(wù)必先滿足使用 Spark / Flink 運行時集群必須已經(jīng)集成 Hadoop官方實測的 Hadoop 版本為 2.x。如果集群未集成 HadoopFTP 文件系統(tǒng)適配層將無法加載作業(yè)會啟動失敗。使用 SeaTunnel Engine 運行時無需額外操作下載并安裝 SeaTunnel Engine 時已自動集成 Hadoop 相關(guān) jar?？赏ㄟ^檢查${SEATUNNEL_HOME}/lib目錄下的 jar 包來確認(rèn)。這一前提在源碼中也得到印證FtpFileSink 通過 Hadoop 配置體系來橋接 FTP 協(xié)議見下文“FTP 文件系統(tǒng)適配”本質(zhì)上仍依賴 Hadoop FileSystem 抽象。核心特性Key features特性說明exactly-once默認(rèn)通過兩階段提交2PC保證exactly-once詳見 connector-v2-features文件格式text、csv、parquet、orc、json、excel、xml、binary 共 8 種均完整支持其中 exactly-once 承諾的落地實現(xiàn)可追溯到 BaseFileSink它實現(xiàn)了SeaTunnelSinkSeaTunnelRow, FileSinkState, FileCommitInfo, FileAggregatedCommitInfo泛型簽名分別提供了createWriter、restoreWriter從 checkpoint 狀態(tài)恢復(fù)寫以及createAggregatedCommitter返回FileSinkAggregatedCommitter這正是 2PC 中預(yù)提交與最終提交的載體詳見下文“源碼級原理”。完整 Options 參數(shù)表下表完整列出官方文檔定義的連接器參數(shù)。標(biāo)注 “Required yes” 的 6 個參數(shù)host、port、user、password、path、tmp_path在配置缺失時連接器會直接拋出配置校驗異常。名稱類型是否必填默認(rèn)值說明hoststringyes-FTP 服務(wù)器地址portintyes-FTP 服務(wù)器端口userstringyes-FTP 用戶名passwordstringyes-FTP 密碼pathstringyes-目標(biāo)目錄路徑tmp_pathstringyes/tmp/seatunnel結(jié)果文件先寫入臨時目錄再通過mv提交到目標(biāo)目錄必須是 FTP 目錄connection_modestringnoactive_localFTP 連接模式custom_filenamebooleannofalse是否自定義文件名file_name_expressionstringno${transactionId}僅當(dāng) custom_filename 為 true 時生效filename_time_formatstringnoyyyy.MM.dd僅當(dāng) custom_filename 為 true 時生效file_format_typestringnocsv輸出文件格式field_delimiterstringno\001僅 text 格式生效列分隔符row_delimiterstringno\n僅 text 格式生效行分隔符have_partitionbooleannofalse是否開啟分區(qū)處理partition_byarrayno-僅 have_partition 為 true 時生效按所選字段分區(qū)partition_dir_expressionstringno${k0}${v0}/${k1}${v1}/.../${kn}${vn}/僅 have_partition 為 true 時生效is_partition_field_write_in_filebooleannofalse僅 have_partition 為 true 時生效sink_columnsarrayno空為空時寫入全部字段順序決定文件實際寫入順序is_enable_transactionbooleannotrue是否開啟事務(wù)保護(hù)batch_sizeintno1000000單個文件最大行數(shù)compress_codecstringnonone壓縮編解碼器common-optionsobjectno-Sink 公共參數(shù)max_rows_in_memoryintno-僅 excel 格式生效內(nèi)存中緩存的最大數(shù)據(jù)條數(shù)sheet_namestringnoSheet${隨機(jī)數(shù)}僅 excel 格式生效工作簿工作表名xml_root_tagstringnoRECORDS僅 xml 格式生效XML 根元素標(biāo)簽xml_row_tagstringnoRECORD僅 xml 格式生效XML 數(shù)據(jù)行標(biāo)簽xml_use_attr_formatbooleanno-僅 xml 格式生效是否使用標(biāo)簽屬性格式parquet_avro_write_timestamp_as_int96booleannofalse僅 parquet 格式生效parquet_avro_write_fixed_as_int96arrayno-僅 parquet 格式生效encodingstringnoUTF-8僅 json、text、csv、xml 格式生效上表默認(rèn)值均可在 BaseSinkConfig 中逐一找到對應(yīng)常量如DEFAULT_TMP_PATH /tmp/seatunnel、DEFAULT_FILE_NAME_EXPRESSION ${transactionId}、DEFAULT_BATCH_SIZE 1000000參數(shù)定義與文檔完全一致可作為排障時的權(quán)威依據(jù)。連接參數(shù)host / port / user / password / path / tmp_path / connection_mode連接五要素與目標(biāo)路徑host[string]FTP 服務(wù)器主機(jī)地址必填。port[int]FTP 服務(wù)端口必填通常為 21。user[string]FTP 登錄用戶名必填。password[string]FTP 登錄密碼必填。path[string]目標(biāo)寫入目錄必填。最終文件會生成在該目錄下。tmp_path[string]臨時目錄必填默認(rèn)/tmp/seatunnel。寫入機(jī)制是結(jié)果文件先寫進(jìn)臨時目錄完成后用mv命令整體提交到目標(biāo)目錄這保證了目標(biāo)目錄不會出現(xiàn)半成品文件。由于mv發(fā)生在 FTP 服務(wù)器側(cè)tmp_path必須是一個 FTP 可達(dá)的目錄且建議與path位于同一文件系統(tǒng)/掛載范圍內(nèi)。在 FtpFileSink.java 的prepare階段會通過CheckConfigUtil.checkAllExists強(qiáng)制校驗 host、port、user、password 四項path 的校驗由工廠的OptionRule完成任一缺失都會拋出FileConnectorException錯誤碼CONFIG_VALIDATION_FAILED。connection_mode主動/被動連接模式默認(rèn)值active_local可選值active_local、passive_local。對應(yīng)枚舉定義在 FtpConnectionMode其模式字符串與 Apache Commons Net 的連接模式一致active_localFTP 主動模式服務(wù)器主動連接客戶端的數(shù)據(jù)端口passive_localFTP 被動模式客戶端主動連接服務(wù)器數(shù)據(jù)端口。生產(chǎn)環(huán)境中客戶端位于 NAT/防火墻之后時通常需要切換為passive_local否則數(shù)據(jù)通道可能建立失敗。FTP 文件系統(tǒng)適配源碼級FtpFile 連接器通過 Hadoop FileSystem 抽象訪問 FTP 協(xié)議這一橋接邏輯在 FtpConf 中實現(xiàn)文件系統(tǒng)實現(xiàn)類org.apache.seatunnel.connectors.seatunnel.file.ftp.system.SeaTunnelFTPFileSystem協(xié)議 schemeftp默認(rèn)文件系統(tǒng)地址defaultFS由配置拼接為ftp://{host}:{port}認(rèn)證與連接參數(shù)被映射為 Hadoop 配置項寫入hadoopConffs.ftp.user.{host}→ FTP 用戶名fs.ftp.password.{host}→ FTP 密碼fs.ftp.connection.mode→ 連接模式僅在配置了connection_mode時寫入。也就是說只要讓BaseFileSinkWriter拿到這份HadoopConf寫數(shù)據(jù)、建目錄、mv提交等所有文件操作都復(fù)用了文件連接器公共模塊的通用實現(xiàn)FTP 特有邏輯被收斂在配置映射這一層。這也是“先臨時目錄、后 mv 提交”機(jī)制能夠透明工作的原因。文件名與時間格式控制custom_filename [boolean]是否自定義文件名默認(rèn)false。file_name_expression [string]僅當(dāng)custom_filename true時生效。描述要寫入path的文件名表達(dá)式支持變量${now}與${uuid}例如test_${uuid}_${now}${now}表示當(dāng)前時間其格式由filename_time_format定義需要注意若is_enable_transaction true會自動在文件名頭部追加${transactionId}_前綴用于事務(wù)標(biāo)識例如transactionId_20250101_xxxx這類形態(tài)。默認(rèn)表達(dá)式為${transactionId}見 BaseSinkConfig 的DEFAULT_FILE_NAME_EXPRESSION。filename_time_format [string]僅當(dāng)custom_filename true時生效。當(dāng)file_name_expression中包含xxxx-${now}時該參數(shù)指定時間格式默認(rèn)yyyy.MM.dd。常用符號如下符號含義y年YearM月Monthd日Day of monthH小時Hour in day 0-23m分鐘Minute in hours秒Second in minute輸出文件格式與行列分隔file_format_type [string]支持的格式text、csv、parquet、orc、json、excel、xml、binary。默認(rèn)值為csv。請?zhí)貏e注意最終文件名會以格式后綴結(jié)尾其中 text 格式的后綴是txt即file_format_type text時生成xxx.txt文件。field_delimiter [string]行內(nèi)列之間的分隔符僅 text 格式需要。默認(rèn)\001SOH 控制符即\u0001這是 Hadoop 生態(tài) text 文件常見的默認(rèn)列分隔。row_delimiter [string]文件內(nèi)行與行之間的分隔符僅 text 格式需要默認(rèn)\n。歷史上2.3.0-beta 之前存在無法從配置文件解析\t作為分隔符的缺陷相關(guān)修復(fù)可見文檔 Changelog 部分配置時建議直接書寫真實轉(zhuǎn)義字符。encoding [string]僅 json、text、csv、xml 四種格式生效。寫入文件的編碼默認(rèn)UTF-8該參數(shù)最終通過Charset.forName(encoding)解析。壓縮配置compress_codeccompress_codec指定輸出文件的壓縮編解碼器默認(rèn)none。不同格式支持的壓縮算法不同務(wù)必按下表對照使用excel 格式不支持任何壓縮文件格式支持壓縮算法txtlzo、nonejsonlzo、nonecsvlzo、noneorclzo、snappy、lz4、zlib、noneparquetlzo、snappy、lz4、gzip、brotli、zstd、none源碼 BaseSinkConfig 中以TXT_COMPRESS、PARQUET_COMPRESS、ORC_COMPRESS三個單選項分別約束了 txt/json/csv、parquet、orc 的可選壓縮集與文檔表格完全一致同時 FtpFileSinkFactory 的OptionRule也按file_format_type做了條件綁定配置不合法時工廠校驗即會報錯。分區(qū)寫入have_partition [boolean]是否開啟分區(qū)處理默認(rèn)false。partition_by [array]僅當(dāng)have_partition true時生效。按所選字段進(jìn)行分區(qū)支持多字段組合。partition_dir_expression [string]僅當(dāng)have_partition true時生效。指定partition_by后連接器會根據(jù)分區(qū)信息生成對應(yīng)的分區(qū)目錄最終文件落在分區(qū)目錄內(nèi)。默認(rèn)表達(dá)式為${k0}${v0}/${k1}${v1}/.../${kn}${vn}/其中k0是第一個分區(qū)字段名v0是第一個分區(qū)字段的值以此類推。例如按age分區(qū)時典型結(jié)果是age25/這樣的目錄層級。is_partition_field_write_in_file [boolean]僅當(dāng)have_partition true時生效。若為true分區(qū)字段及其值也會寫入數(shù)據(jù)文件內(nèi)部若為false分區(qū)字段只體現(xiàn)在目錄結(jié)構(gòu)中。官方文檔特別舉例如果要產(chǎn)出 Hive 數(shù)據(jù)文件該值應(yīng)設(shè)為falseHive 分區(qū)約定分區(qū)列不在數(shù)據(jù)文件內(nèi)重復(fù)出現(xiàn)。寫入列與事務(wù)保護(hù)sink_columns [array]需要寫入文件的列默認(rèn)值為 Source 或 Transform 輸出的全部列。數(shù)組中字段的順序即文件實際寫入順序可用于調(diào)整列序或裁剪列。is_enable_transaction [boolean]默認(rèn)true。開啟后連接器保證寫入目標(biāo)目錄的數(shù)據(jù)不丟失、不重復(fù)。開啟時文件名頭部會自動追加${transactionId}_前綴。官方文檔明確當(dāng)前只支持true。batch_size [int]單個文件的最大行數(shù)默認(rèn)1000000。在 SeaTunnel Engine 中一個文件的實際行數(shù)由batch_size與checkpoint.interval共同決定若checkpoint.interval足夠大sink writer 會持續(xù)寫入當(dāng)前文件直到行數(shù)超過batch_size才切換新文件若checkpoint.interval較小則每次 checkpoint 觸發(fā)時sink writer 就會創(chuàng)建一個新文件以 checkpoint 邊界切分文件便于與事務(wù)提交對齊。因此想要控制文件大小應(yīng)同時調(diào)大checkpoint.interval并設(shè)置合適的batch_size。Excel / XML / Parquet 專屬參數(shù)max_rows_in_memory [int]僅 excel 格式生效。內(nèi)存中可緩存的最大數(shù)據(jù)條數(shù)用于控制寫 Excel 時的內(nèi)存占用。sheet_name [string]僅 excel 格式生效。寫入的工作簿工作表名默認(rèn)值為Sheet${隨機(jī)數(shù)}。xml_root_tag [string]僅 xml 格式生效。指定 XML 文件的根元素標(biāo)簽名默認(rèn)RECORDS。xml_row_tag [string]僅 xml 格式生效。指定 XML 文件中數(shù)據(jù)行的標(biāo)簽名默認(rèn)RECORD。xml_use_attr_format [boolean]僅 xml 格式生效。是否使用標(biāo)簽屬性格式處理數(shù)據(jù)。parquet_avro_write_timestamp_as_int96 [boolean]僅 parquet 格式生效默認(rèn)false。支持將 timestamp 類型寫為 Parquet INT96。parquet_avro_write_fixed_as_int96 [array]僅 parquet 格式生效。支持將 12 字節(jié)字段寫為 Parquet INT96。完整配置示例示例一text 格式的基礎(chǔ)配置最簡單的 text 落地配置官方文檔原始示例FtpFile { host xxx.xxx.xxx.xxx port 21 user username password password path /data/ftp file_format_type text field_delimiter \t row_delimiter \n sink_columns [name,age] }該配置將上游數(shù)據(jù)以name、age兩列順序?qū)懭?data/ftp下的*.txt文件列間以\t分隔行間以\n分隔未開啟自定義文件名默認(rèn)使用${transactionId}表達(dá)式并帶事務(wù)前綴未開啟分區(qū)。示例二分區(qū) 自定義文件名 列裁剪的組合配置同時啟用have_partition、custom_filename與sink_columns的完整配置官方文檔原始示例FtpFile { host xxx.xxx.xxx.xxx port 21 user username password password path /data/ftp/seatunnel/job1 tmp_path /data/ftp/seatunnel/tmp file_format_type text field_delimiter \t row_delimiter \n have_partition true partition_by [age] partition_dir_expression ${k0}${v0} is_partition_field_write_in_file true custom_filename true file_name_expression ${transactionId}_${now} sink_columns [name,age] filename_time_format yyyy.MM.dd }行為推演結(jié)合前文參數(shù)語義數(shù)據(jù)先寫入臨時目錄/data/ftp/seatunnel/tmp提交時mv到目標(biāo)目錄/data/ftp/seatunnel/job1按age字段分區(qū)分區(qū)目錄表達(dá)式為${k0}${v0}例如age25/is_partition_field_write_in_file true分區(qū)列age也會寫入數(shù)據(jù)文件文件名形如${transactionId}_2025.01.01.txt${now}按yyyy.MM.dd格式化text 后綴為 txt僅寫入name、age兩列順序按數(shù)組聲明。源碼級工作原理與驗證Sink 主類與配置校驗FtpFileSink 通過AutoService(SeaTunnelSink.class)注冊插件名為FileSystemType.FTP對應(yīng)的字符串。其prepare方法強(qiáng)制校驗 host/port/user/password隨后調(diào)用FtpConf.buildWithConfig構(gòu)造 Hadoop 配置其余邏輯全部繼承自 BaseFileSink。2PC exactly-once 的實現(xiàn)路徑從 BaseFileSink 的接口實現(xiàn)可以看出完整的提交鏈路createWriter/restoreWriter創(chuàng)建或恢復(fù)BaseFileSinkWriter寫入發(fā)生在臨時路徑createAggregatedCommitter返回FileSinkAggregatedCommitter在 checkpoint 提交階段統(tǒng)一把各 writer 的臨時文件mv到目標(biāo)目錄三個序列化器FileCommitInfo、FileAggregatedCommitInfo、FileSinkState保證 checkpoint 狀態(tài)可持久化、可恢復(fù)。這套“寫臨時文件 → checkpoint 預(yù)提交 → 聚合提交mv 到目標(biāo)目錄”的機(jī)制正是“數(shù)據(jù)不丟失、不重復(fù)”與文件名帶transactionId前綴的原因。文檔中的“Next version”Changelog 提到修復(fù)了“從 states 恢復(fù) writer 時獲取 transaction 直接失敗”的問題正是這條鏈路中的狀態(tài)恢復(fù)環(huán)節(jié)。工廠參數(shù)約束與單元測試FtpFileSinkFactory 通過OptionRule聲明了參數(shù)依賴關(guān)系path、host、port、user、password為必填field_delimiter、row_delimiter、壓縮等參數(shù)僅在對應(yīng)file_format_type下生效file_name_expression、filename_time_format僅在custom_filename true時生效partition_by等僅在have_partition true時生效。倉庫中的單元測試 FtpFileFactoryTest 對 Sink 與 Source 工廠的optionRule()均做了非空斷言確保參數(shù)規(guī)則可用。Sink 與 Source 的能力差異注意同一 FTP 模塊下的 FtpFileSource 在讀取側(cè)有明確限制Ftp 文件 Source 僅支持讀取 text、csv、json 三種格式配置為 orc、parquet 會直接拋出ILLEGAL_ARGUMENT異常而本篇文章所講的 FtpFileSink則支持 8 種格式的寫入。在規(guī)劃“FTP 讀 → FTP 寫”的自同步任務(wù)時需注意兩側(cè)格式能力并不對稱。版本演進(jìn) Changelog2.2.0-beta2022-09-26新增 Ftp File Sink 連接器。2.3.0-beta2022-10-20三處缺陷修復(fù)——Windows 環(huán)境下路徑不正確PR 2980文件系統(tǒng)獲取報錯PR 3117配置文件無法解析\t作為分隔符PR 3083。Next version缺陷修復(fù)PR 3258上游字段為 null 時拋 NullPointerExceptionsink 列映射失敗從 states 恢復(fù) writer 時直接獲取 transaction 失敗能力增強(qiáng)支持為每個文件設(shè)置 batch_sizePR 3625支持文件壓縮PR 3899。常見問題速查現(xiàn)象排查方向作業(yè)啟動報配置校驗失敗檢查 host、port、user、password、path 是否齊全對應(yīng)FtpFileSink.prepare與OptionRule的必填聲明Spark/Flink 下無法寫入確認(rèn)集群已集成 Hadoop 2.xFTP 文件系統(tǒng)適配依賴 Hadoop FileSystem 抽象數(shù)據(jù)通道建立失敗客戶端處于 NAT/防火墻后時將connection_mode改為passive_local文件一直不落到目標(biāo)目錄確認(rèn)tmp_path為 FTP 可達(dá)目錄mv提交依賴該目錄可寫同時理解文件切分由batch_size與checkpoint.interval共同決定目標(biāo)目錄出現(xiàn)未知前綴文件名is_enable_transaction true時文件名自動帶${transactionId}_前綴屬預(yù)期行為寫 Hive 數(shù)據(jù)文件列重復(fù)分區(qū)寫入時確認(rèn)is_partition_field_write_in_file false壓縮參數(shù)報錯對照“壓縮配置”表格excel 不支持壓縮orc/parquet 各有允許的算法白名單更多 Sink 公共參數(shù)如date_format、datetime_format、time_format、表頭寫入等可查閱 Sink Common OptionsFtpFile 專屬配置之外的能力均可復(fù)用公共參數(shù)體系。贊分享數(shù)據(jù)工程大數(shù)據(jù)批處理流處理【免費下載鏈接】seatunnelSeaTunnel is a next-generation super high-performance, distributed, massive data integration tool.項目地址https://gitcode.com/gh_mirrors/sea/seatunnel點擊查看免費下載相關(guān)推薦SeaTunnel FtpFile Sink 連接器實戰(zhàn)指南將數(shù)據(jù)可靠寫入 FTP/SFTPSeaTunnel FtpFile Sink 連接器實戰(zhàn)指南將數(shù)據(jù)可靠寫入 FTP/SFTP 本文是 SeaTunnel 文件類 Sink 連接器系列中的 F數(shù)據(jù)集成ETL大數(shù)據(jù)批處理流處理變更數(shù)據(jù)捕獲SeaTunnel FtpFile Sink Connector 使用指南將數(shù)據(jù)輸出到 FTP 服務(wù)器SeaTunnel FtpFile Sink Connector 使用指南將數(shù)據(jù)輸出到 FTP 服務(wù)器 SeaTunnel 的 FtpFile Sink 插件數(shù)據(jù)集成ETL大數(shù)據(jù)批處理流處理變更數(shù)據(jù)捕獲Apache SeaTunnel SLS Sink 連接器將 SeaTunnel 行數(shù)據(jù)寫入阿里云日志服務(wù)的實現(xiàn)與配置指南Apache SeaTunnel SLS Sink 連接器將 SeaTunnel 行數(shù)據(jù)寫入阿里云日志服務(wù)的實現(xiàn)與配置指南 本文基于 SeaTunnel 官方數(shù)據(jù)集成ETL大數(shù)據(jù)批處理流處理變更數(shù)據(jù)捕獲上一篇Isar數(shù)據(jù)庫最佳實踐構(gòu)建高性能Flutter應(yīng)用的25個技巧下一篇CANN/asc-devkit Sign接口緩沖區(qū)因子大小獲取創(chuàng)作聲明：本文部分內(nèi)容由AI輔助生成（AIGC），僅供參考 SeaTunnel FtpFile Sink 連接器完全指南：將數(shù)據(jù)寫入 FTP 服務(wù)器的配置、參數(shù)與實現(xiàn)原理
數(shù)據(jù)工程大數(shù)據(jù)批處理流處理【免費下載鏈接】seatunnelSeaTunnel is a next-generation super high-performance, distributed, massive data integration tool.項目地址https://gitcode.com/gh_mirrors/sea/seatunnel點擊查看免費下載FtpFile Sink 是 Apache SeaTunnel 內(nèi)置的文件型輸出連接器負(fù)責(zé)把上游 Source 或 Transform 產(chǎn)出的數(shù)據(jù)以 text、csv、parquet、orc、json、excel、xml、binary 等格式寫入 FTP 服務(wù)器指定目錄。本文以官方文檔 docs/en/connector-v2/sink/FtpFile.md 為主體結(jié)合倉庫中connector-file-ftp與connector-file-base模塊源碼完整講解該連接器的全部配置參數(shù)、運行環(huán)境要求、分區(qū)與事務(wù)機(jī)制、完整配置示例以及 2PC 提交與 FTP 文件系統(tǒng)適配的底層實現(xiàn)。讀完本文你將能夠獨立完成一個從任意 Source 到 FTP 的 SeaTunnel 作業(yè)的配置、調(diào)優(yōu)與排障。連接器簡介與適用場景FtpFile 連接器的職責(zé)非常聚焦Output data to Ftp——將 SeaTunnel 管道中的SeaTunnelRow數(shù)據(jù)流式寫入 FTP 服務(wù)器。它與同模塊的 FtpFileSource 形成一對讀寫組合適用于數(shù)據(jù)落地 FTP、與外部系統(tǒng)通過 FTP 交換文件的場景。從源碼看連接器插件名來自 FileSystemType.FTPSink 側(cè)實現(xiàn)為 FtpFileSink它繼承自文件連接器公共基類BaseFileSink因此所有文件型 Sink 的通用能力分區(qū)、事務(wù)、格式、壓縮它都天然具備。運行環(huán)境前提官方文檔對運行環(huán)境有明確提示務(wù)必先滿足使用 Spark / Flink 運行時集群必須已經(jīng)集成 Hadoop官方實測的 Hadoop 版本為 2.x。如果集群未集成 HadoopFTP 文件系統(tǒng)適配層將無法加載作業(yè)會啟動失敗。使用 SeaTunnel Engine 運行時無需額外操作下載并安裝 SeaTunnel Engine 時已自動集成 Hadoop 相關(guān) jar?？赏ㄟ^檢查${SEATUNNEL_HOME}/lib目錄下的 jar 包來確認(rèn)。這一前提在源碼中也得到印證FtpFileSink 通過 Hadoop 配置體系來橋接 FTP 協(xié)議見下文“FTP 文件系統(tǒng)適配”本質(zhì)上仍依賴 Hadoop FileSystem 抽象。核心特性Key features特性說明exactly-once默認(rèn)通過兩階段提交2PC保證exactly-once詳見 connector-v2-features文件格式text、csv、parquet、orc、json、excel、xml、binary 共 8 種均完整支持其中 exactly-once 承諾的落地實現(xiàn)可追溯到 BaseFileSink它實現(xiàn)了SeaTunnelSinkSeaTunnelRow, FileSinkState, FileCommitInfo, FileAggregatedCommitInfo泛型簽名分別提供了createWriter、restoreWriter從 checkpoint 狀態(tài)恢復(fù)寫以及createAggregatedCommitter返回FileSinkAggregatedCommitter這正是 2PC 中預(yù)提交與最終提交的載體詳見下文“源碼級原理”。完整 Options 參數(shù)表下表完整列出官方文檔定義的連接器參數(shù)。標(biāo)注 “Required yes” 的 6 個參數(shù)host、port、user、password、path、tmp_path在配置缺失時連接器會直接拋出配置校驗異常。名稱類型是否必填默認(rèn)值說明hoststringyes-FTP 服務(wù)器地址portintyes-FTP 服務(wù)器端口userstringyes-FTP 用戶名passwordstringyes-FTP 密碼pathstringyes-目標(biāo)目錄路徑tmp_pathstringyes/tmp/seatunnel結(jié)果文件先寫入臨時目錄再通過mv提交到目標(biāo)目錄必須是 FTP 目錄connection_modestringnoactive_localFTP 連接模式custom_filenamebooleannofalse是否自定義文件名file_name_expressionstringno${transactionId}僅當(dāng) custom_filename 為 true 時生效filename_time_formatstringnoyyyy.MM.dd僅當(dāng) custom_filename 為 true 時生效file_format_typestringnocsv輸出文件格式field_delimiterstringno\001僅 text 格式生效列分隔符row_delimiterstringno\n僅 text 格式生效行分隔符have_partitionbooleannofalse是否開啟分區(qū)處理partition_byarrayno-僅 have_partition 為 true 時生效按所選字段分區(qū)partition_dir_expressionstringno${k0}${v0}/${k1}${v1}/.../${kn}${vn}/僅 have_partition 為 true 時生效is_partition_field_write_in_filebooleannofalse僅 have_partition 為 true 時生效sink_columnsarrayno空為空時寫入全部字段順序決定文件實際寫入順序is_enable_transactionbooleannotrue是否開啟事務(wù)保護(hù)batch_sizeintno1000000單個文件最大行數(shù)compress_codecstringnonone壓縮編解碼器common-optionsobjectno-Sink 公共參數(shù)max_rows_in_memoryintno-僅 excel 格式生效內(nèi)存中緩存的最大數(shù)據(jù)條數(shù)sheet_namestringnoSheet${隨機(jī)數(shù)}僅 excel 格式生效工作簿工作表名xml_root_tagstringnoRECORDS僅 xml 格式生效XML 根元素標(biāo)簽xml_row_tagstringnoRECORD僅 xml 格式生效XML 數(shù)據(jù)行標(biāo)簽xml_use_attr_formatbooleanno-僅 xml 格式生效是否使用標(biāo)簽屬性格式parquet_avro_write_timestamp_as_int96booleannofalse僅 parquet 格式生效parquet_avro_write_fixed_as_int96arrayno-僅 parquet 格式生效encodingstringnoUTF-8僅 json、text、csv、xml 格式生效上表默認(rèn)值均可在 BaseSinkConfig 中逐一找到對應(yīng)常量如DEFAULT_TMP_PATH /tmp/seatunnel、DEFAULT_FILE_NAME_EXPRESSION ${transactionId}、DEFAULT_BATCH_SIZE 1000000參數(shù)定義與文檔完全一致可作為排障時的權(quán)威依據(jù)。連接參數(shù)host / port / user / password / path / tmp_path / connection_mode連接五要素與目標(biāo)路徑host[string]FTP 服務(wù)器主機(jī)地址必填。port[int]FTP 服務(wù)端口必填通常為 21。user[string]FTP 登錄用戶名必填。password[string]FTP 登錄密碼必填。path[string]目標(biāo)寫入目錄必填。最終文件會生成在該目錄下。tmp_path[string]臨時目錄必填默認(rèn)/tmp/seatunnel。寫入機(jī)制是結(jié)果文件先寫進(jìn)臨時目錄完成后用mv命令整體提交到目標(biāo)目錄這保證了目標(biāo)目錄不會出現(xiàn)半成品文件。由于mv發(fā)生在 FTP 服務(wù)器側(cè)tmp_path必須是一個 FTP 可達(dá)的目錄且建議與path位于同一文件系統(tǒng)/掛載范圍內(nèi)。在 FtpFileSink.java 的prepare階段會通過CheckConfigUtil.checkAllExists強(qiáng)制校驗 host、port、user、password 四項path 的校驗由工廠的OptionRule完成任一缺失都會拋出FileConnectorException錯誤碼CONFIG_VALIDATION_FAILED。connection_mode主動/被動連接模式默認(rèn)值active_local可選值active_local、passive_local。對應(yīng)枚舉定義在 FtpConnectionMode其模式字符串與 Apache Commons Net 的連接模式一致active_localFTP 主動模式服務(wù)器主動連接客戶端的數(shù)據(jù)端口passive_localFTP 被動模式客戶端主動連接服務(wù)器數(shù)據(jù)端口。生產(chǎn)環(huán)境中客戶端位于 NAT/防火墻之后時通常需要切換為passive_local否則數(shù)據(jù)通道可能建立失敗。FTP 文件系統(tǒng)適配源碼級FtpFile 連接器通過 Hadoop FileSystem 抽象訪問 FTP 協(xié)議這一橋接邏輯在 FtpConf 中實現(xiàn)文件系統(tǒng)實現(xiàn)類org.apache.seatunnel.connectors.seatunnel.file.ftp.system.SeaTunnelFTPFileSystem協(xié)議 schemeftp默認(rèn)文件系統(tǒng)地址defaultFS由配置拼接為ftp://{host}:{port}認(rèn)證與連接參數(shù)被映射為 Hadoop 配置項寫入hadoopConffs.ftp.user.{host}→ FTP 用戶名fs.ftp.password.{host}→ FTP 密碼fs.ftp.connection.mode→ 連接模式僅在配置了connection_mode時寫入。也就是說只要讓BaseFileSinkWriter拿到這份HadoopConf寫數(shù)據(jù)、建目錄、mv提交等所有文件操作都復(fù)用了文件連接器公共模塊的通用實現(xiàn)FTP 特有邏輯被收斂在配置映射這一層。這也是“先臨時目錄、后 mv 提交”機(jī)制能夠透明工作的原因。文件名與時間格式控制custom_filename [boolean]是否自定義文件名默認(rèn)false。file_name_expression [string]僅當(dāng)custom_filename true時生效。描述要寫入path的文件名表達(dá)式支持變量${now}與${uuid}例如test_${uuid}_${now}${now}表示當(dāng)前時間其格式由filename_time_format定義需要注意若is_enable_transaction true會自動在文件名頭部追加${transactionId}_前綴用于事務(wù)標(biāo)識例如transactionId_20250101_xxxx這類形態(tài)。默認(rèn)表達(dá)式為${transactionId}見 BaseSinkConfig 的DEFAULT_FILE_NAME_EXPRESSION。filename_time_format [string]僅當(dāng)custom_filename true時生效。當(dāng)file_name_expression中包含xxxx-${now}時該參數(shù)指定時間格式默認(rèn)yyyy.MM.dd。常用符號如下符號含義y年YearM月Monthd日Day of monthH小時Hour in day 0-23m分鐘Minute in hours秒Second in minute輸出文件格式與行列分隔file_format_type [string]支持的格式text、csv、parquet、orc、json、excel、xml、binary。默認(rèn)值為csv。請?zhí)貏e注意最終文件名會以格式后綴結(jié)尾其中 text 格式的后綴是txt即file_format_type text時生成xxx.txt文件。field_delimiter [string]行內(nèi)列之間的分隔符僅 text 格式需要。默認(rèn)\001SOH 控制符即\u0001這是 Hadoop 生態(tài) text 文件常見的默認(rèn)列分隔。row_delimiter [string]文件內(nèi)行與行之間的分隔符僅 text 格式需要默認(rèn)\n。歷史上2.3.0-beta 之前存在無法從配置文件解析\t作為分隔符的缺陷相關(guān)修復(fù)可見文檔 Changelog 部分配置時建議直接書寫真實轉(zhuǎn)義字符。encoding [string]僅 json、text、csv、xml 四種格式生效。寫入文件的編碼默認(rèn)UTF-8該參數(shù)最終通過Charset.forName(encoding)解析。壓縮配置compress_codeccompress_codec指定輸出文件的壓縮編解碼器默認(rèn)none。不同格式支持的壓縮算法不同務(wù)必按下表對照使用excel 格式不支持任何壓縮文件格式支持壓縮算法txtlzo、nonejsonlzo、nonecsvlzo、noneorclzo、snappy、lz4、zlib、noneparquetlzo、snappy、lz4、gzip、brotli、zstd、none源碼 BaseSinkConfig 中以TXT_COMPRESS、PARQUET_COMPRESS、ORC_COMPRESS三個單選項分別約束了 txt/json/csv、parquet、orc 的可選壓縮集與文檔表格完全一致同時 FtpFileSinkFactory 的OptionRule也按file_format_type做了條件綁定配置不合法時工廠校驗即會報錯。分區(qū)寫入have_partition [boolean]是否開啟分區(qū)處理默認(rèn)false。partition_by [array]僅當(dāng)have_partition true時生效。按所選字段進(jìn)行分區(qū)支持多字段組合。partition_dir_expression [string]僅當(dāng)have_partition true時生效。指定partition_by后連接器會根據(jù)分區(qū)信息生成對應(yīng)的分區(qū)目錄最終文件落在分區(qū)目錄內(nèi)。默認(rèn)表達(dá)式為${k0}${v0}/${k1}${v1}/.../${kn}${vn}/其中k0是第一個分區(qū)字段名v0是第一個分區(qū)字段的值以此類推。例如按age分區(qū)時典型結(jié)果是age25/這樣的目錄層級。is_partition_field_write_in_file [boolean]僅當(dāng)have_partition true時生效。若為true分區(qū)字段及其值也會寫入數(shù)據(jù)文件內(nèi)部若為false分區(qū)字段只體現(xiàn)在目錄結(jié)構(gòu)中。官方文檔特別舉例如果要產(chǎn)出 Hive 數(shù)據(jù)文件該值應(yīng)設(shè)為falseHive 分區(qū)約定分區(qū)列不在數(shù)據(jù)文件內(nèi)重復(fù)出現(xiàn)。寫入列與事務(wù)保護(hù)sink_columns [array]需要寫入文件的列默認(rèn)值為 Source 或 Transform 輸出的全部列。數(shù)組中字段的順序即文件實際寫入順序可用于調(diào)整列序或裁剪列。is_enable_transaction [boolean]默認(rèn)true。開啟后連接器保證寫入目標(biāo)目錄的數(shù)據(jù)不丟失、不重復(fù)。開啟時文件名頭部會自動追加${transactionId}_前綴。官方文檔明確當(dāng)前只支持true。batch_size [int]單個文件的最大行數(shù)默認(rèn)1000000。在 SeaTunnel Engine 中一個文件的實際行數(shù)由batch_size與checkpoint.interval共同決定若checkpoint.interval足夠大sink writer 會持續(xù)寫入當(dāng)前文件直到行數(shù)超過batch_size才切換新文件若checkpoint.interval較小則每次 checkpoint 觸發(fā)時sink writer 就會創(chuàng)建一個新文件以 checkpoint 邊界切分文件便于與事務(wù)提交對齊。因此想要控制文件大小應(yīng)同時調(diào)大checkpoint.interval并設(shè)置合適的batch_size。Excel / XML / Parquet 專屬參數(shù)max_rows_in_memory [int]僅 excel 格式生效。內(nèi)存中可緩存的最大數(shù)據(jù)條數(shù)用于控制寫 Excel 時的內(nèi)存占用。sheet_name [string]僅 excel 格式生效。寫入的工作簿工作表名默認(rèn)值為Sheet${隨機(jī)數(shù)}。xml_root_tag [string]僅 xml 格式生效。指定 XML 文件的根元素標(biāo)簽名默認(rèn)RECORDS。xml_row_tag [string]僅 xml 格式生效。指定 XML 文件中數(shù)據(jù)行的標(biāo)簽名默認(rèn)RECORD。xml_use_attr_format [boolean]僅 xml 格式生效。是否使用標(biāo)簽屬性格式處理數(shù)據(jù)。parquet_avro_write_timestamp_as_int96 [boolean]僅 parquet 格式生效默認(rèn)false。支持將 timestamp 類型寫為 Parquet INT96。parquet_avro_write_fixed_as_int96 [array]僅 parquet 格式生效。支持將 12 字節(jié)字段寫為 Parquet INT96。完整配置示例示例一text 格式的基礎(chǔ)配置最簡單的 text 落地配置官方文檔原始示例FtpFile { host xxx.xxx.xxx.xxx port 21 user username password password path /data/ftp file_format_type text field_delimiter \t row_delimiter \n sink_columns [name,age] }該配置將上游數(shù)據(jù)以name、age兩列順序?qū)懭?data/ftp下的*.txt文件列間以\t分隔行間以\n分隔未開啟自定義文件名默認(rèn)使用${transactionId}表達(dá)式并帶事務(wù)前綴未開啟分區(qū)。示例二分區(qū) 自定義文件名 列裁剪的組合配置同時啟用have_partition、custom_filename與sink_columns的完整配置官方文檔原始示例FtpFile { host xxx.xxx.xxx.xxx port 21 user username password password path /data/ftp/seatunnel/job1 tmp_path /data/ftp/seatunnel/tmp file_format_type text field_delimiter \t row_delimiter \n have_partition true partition_by [age] partition_dir_expression ${k0}${v0} is_partition_field_write_in_file true custom_filename true file_name_expression ${transactionId}_${now} sink_columns [name,age] filename_time_format yyyy.MM.dd }行為推演結(jié)合前文參數(shù)語義數(shù)據(jù)先寫入臨時目錄/data/ftp/seatunnel/tmp提交時mv到目標(biāo)目錄/data/ftp/seatunnel/job1按age字段分區(qū)分區(qū)目錄表達(dá)式為${k0}${v0}例如age25/is_partition_field_write_in_file true分區(qū)列age也會寫入數(shù)據(jù)文件文件名形如${transactionId}_2025.01.01.txt${now}按yyyy.MM.dd格式化text 后綴為 txt僅寫入name、age兩列順序按數(shù)組聲明。源碼級工作原理與驗證Sink 主類與配置校驗FtpFileSink 通過AutoService(SeaTunnelSink.class)注冊插件名為FileSystemType.FTP對應(yīng)的字符串。其prepare方法強(qiáng)制校驗 host/port/user/password隨后調(diào)用FtpConf.buildWithConfig構(gòu)造 Hadoop 配置其余邏輯全部繼承自 BaseFileSink。2PC exactly-once 的實現(xiàn)路徑從 BaseFileSink 的接口實現(xiàn)可以看出完整的提交鏈路createWriter/restoreWriter創(chuàng)建或恢復(fù)BaseFileSinkWriter寫入發(fā)生在臨時路徑createAggregatedCommitter返回FileSinkAggregatedCommitter在 checkpoint 提交階段統(tǒng)一把各 writer 的臨時文件mv到目標(biāo)目錄三個序列化器FileCommitInfo、FileAggregatedCommitInfo、FileSinkState保證 checkpoint 狀態(tài)可持久化、可恢復(fù)。這套“寫臨時文件 → checkpoint 預(yù)提交 → 聚合提交mv 到目標(biāo)目錄”的機(jī)制正是“數(shù)據(jù)不丟失、不重復(fù)”與文件名帶transactionId前綴的原因。文檔中的“Next version”Changelog 提到修復(fù)了“從 states 恢復(fù) writer 時獲取 transaction 直接失敗”的問題正是這條鏈路中的狀態(tài)恢復(fù)環(huán)節(jié)。工廠參數(shù)約束與單元測試FtpFileSinkFactory 通過OptionRule聲明了參數(shù)依賴關(guān)系path、host、port、user、password為必填field_delimiter、row_delimiter、壓縮等參數(shù)僅在對應(yīng)file_format_type下生效file_name_expression、filename_time_format僅在custom_filename true時生效partition_by等僅在have_partition true時生效。倉庫中的單元測試 FtpFileFactoryTest 對 Sink 與 Source 工廠的optionRule()均做了非空斷言確保參數(shù)規(guī)則可用。Sink 與 Source 的能力差異注意同一 FTP 模塊下的 FtpFileSource 在讀取側(cè)有明確限制Ftp 文件 Source 僅支持讀取 text、csv、json 三種格式配置為 orc、parquet 會直接拋出ILLEGAL_ARGUMENT異常而本篇文章所講的 FtpFileSink則支持 8 種格式的寫入。在規(guī)劃“FTP 讀 → FTP 寫”的自同步任務(wù)時需注意兩側(cè)格式能力并不對稱。版本演進(jìn) Changelog2.2.0-beta2022-09-26新增 Ftp File Sink 連接器。2.3.0-beta2022-10-20三處缺陷修復(fù)——Windows 環(huán)境下路徑不正確PR 2980文件系統(tǒng)獲取報錯PR 3117配置文件無法解析\t作為分隔符PR 3083。Next version缺陷修復(fù)PR 3258上游字段為 null 時拋 NullPointerExceptionsink 列映射失敗從 states 恢復(fù) writer 時直接獲取 transaction 失敗能力增強(qiáng)支持為每個文件設(shè)置 batch_sizePR 3625支持文件壓縮PR 3899。常見問題速查現(xiàn)象排查方向作業(yè)啟動報配置校驗失敗檢查 host、port、user、password、path 是否齊全對應(yīng)FtpFileSink.prepare與OptionRule的必填聲明Spark/Flink 下無法寫入確認(rèn)集群已集成 Hadoop 2.xFTP 文件系統(tǒng)適配依賴 Hadoop FileSystem 抽象數(shù)據(jù)通道建立失敗客戶端處于 NAT/防火墻后時將connection_mode改為passive_local文件一直不落到目標(biāo)目錄確認(rèn)tmp_path為 FTP 可達(dá)目錄mv提交依賴該目錄可寫同時理解文件切分由batch_size與checkpoint.interval共同決定目標(biāo)目錄出現(xiàn)未知前綴文件名is_enable_transaction true時文件名自動帶${transactionId}_前綴屬預(yù)期行為寫 Hive 數(shù)據(jù)文件列重復(fù)分區(qū)寫入時確認(rèn)is_partition_field_write_in_file false壓縮參數(shù)報錯對照“壓縮配置”表格excel 不支持壓縮orc/parquet 各有允許的算法白名單更多 Sink 公共參數(shù)如date_format、datetime_format、time_format、表頭寫入等可查閱 Sink Common OptionsFtpFile 專屬配置之外的能力均可復(fù)用公共參數(shù)體系。贊分享數(shù)據(jù)工程大數(shù)據(jù)批處理流處理【免費下載鏈接】seatunnelSeaTunnel is a next-generation super high-performance, distributed, massive data integration tool.項目地址https://gitcode.com/gh_mirrors/sea/seatunnel點擊查看免費下載相關(guān)推薦SeaTunnel FtpFile Sink 連接器實戰(zhàn)指南將數(shù)據(jù)可靠寫入 FTP/SFTPSeaTunnel FtpFile Sink 連接器實戰(zhàn)指南將數(shù)據(jù)可靠寫入 FTP/SFTP 本文是 SeaTunnel 文件類 Sink 連接器系列中的 F數(shù)據(jù)集成ETL大數(shù)據(jù)批處理流處理變更數(shù)據(jù)捕獲SeaTunnel FtpFile Sink Connector 使用指南將數(shù)據(jù)輸出到 FTP 服務(wù)器SeaTunnel FtpFile Sink Connector 使用指南將數(shù)據(jù)輸出到 FTP 服務(wù)器 SeaTunnel 的 FtpFile Sink 插件數(shù)據(jù)集成ETL大數(shù)據(jù)批處理流處理變更數(shù)據(jù)捕獲Apache SeaTunnel SLS Sink 連接器將 SeaTunnel 行數(shù)據(jù)寫入阿里云日志服務(wù)的實現(xiàn)與配置指南Apache SeaTunnel SLS Sink 連接器將 SeaTunnel 行數(shù)據(jù)寫入阿里云日志服務(wù)的實現(xiàn)與配置指南 本文基于 SeaTunnel 官方數(shù)據(jù)集成ETL大數(shù)據(jù)批處理流處理變更數(shù)據(jù)捕獲上一篇Isar數(shù)據(jù)庫最佳實踐構(gòu)建高性能Flutter應(yīng)用的25個技巧下一篇CANN/asc-devkit Sign接口緩沖區(qū)因子大小獲取創(chuàng)作聲明：本文部分內(nèi)容由AI輔助生成（AIGC），僅供參考