Apache SeaTunnel BosFile Sink 深入解析基于 HDFS SDK 向百度智能云 BOS 对象存储写入数据【免费下载链接】seatunnelSeaTunnel is a multimodal, high-performance, distributed, massive data integration tool.项目地址: https://gitcode.com/GitHub_Trending/se/seatunnelBosFile 是 SeaTunnel 提供的对象存储文件写入连接器它通过百度 BOS HDFS SDK 将同步任务的结果写入百度智能云 BOSBaidu Object Storage桶中的指定路径支持 text、csv、parquet、orc、json、excel、xml、binary 等多种文件格式并默认基于 2PC 事务机制保证 exactly-once 语义。读完本文你可以掌握 BosFile sink 的完整参数配置、自定义文件名与分区目录的写法、基于临时目录 重命名的提交原理以及部署bos-hdfs-sdk运行依赖的注意事项。连接器概览引擎支持与运行前提BosFile sink 支持以下计算引擎SeaTunnel ZetaSparkFlink在部署前需要特别注意依赖与 Hadoop 版本前提引自 官方文档若使用Spark / Flink必须确保集群已集成 Hadoop当前经过测试的 Hadoop 版本为2.x若使用SeaTunnel Zeta 引擎Hadoop 相关 jar 已打包在${SEATUNNEL_HOME}/lib下无需额外处理无论哪种引擎都必须将bos-hdfs-sdk 1.0.4-community放入${SEATUNNEL_HOME}/lib。该 jar 并未发布到 Maven CentralSeaTunnel 也不在 pom 中声明它只能作为运行时依赖手工安装。为什么强调 1.0.4 版本下限仓库中专门维护的 SDK 说明文件 connector-file-bos/lib/README.md 给出了明确解释bos-hdfs-sdkisnot published to Maven Central. SeaTunnel does not declare it as a Maven dependency; users must install the jar atruntimeonly.Use1.0.4. SDK 1.0.3 always callsheadBucketduring FileSystem init and ignoresfs.bos.bucket.hierarchyfalse, which fails when the AK/SK lacksHeadBucketpermission.也就是说 1.0.3 及更早版本在 FileSystem 初始化时总会调用headBucket并忽略fs.bos.bucket.hierarchyfalse配置当 AK/SK 缺少HeadBucket权限时任务会直接失败。下载与安装步骤以该文件及 BosFile source 文档 的说明为准。实现原理把 BOS 当作 Hadoop FileSystemBosFile sink 并不是直接使用 BOS 的 REST API 写对象而是复用了 SeaTunnel 文件连接器体系的Hadoop FileSystem 抽象层从而与其他文件系统S3、COS、OSS 等共享同一套写入、切分、事务提交逻辑。核心实现类链路如下BosFileSinkFactory.java插件工厂factoryIdentifier()返回BosFile对应任务配置中sink { BosFile { ... } }的插件名并声明了OptionRule校验规则——path、bucket、access_key、secret_key、endpoint为必填项其余参数按文件类型、custom_filename、have_partition等条件按需生效BosFileSink.javasink 入口继承通用文件写入基类BaseFileSink仅通过覆写initHadoopConf()注入 BOS 专属的 Hadoop 配置BosConf.java将连接器参数翻译为 HDFS 客户端配置。从 BosConf.java 的源码可以看到关键映射private static final String HDFS_IMPL org.apache.hadoop.fs.bos.BaiduBosFileSystem; private static final String SCHEMA bos; private static final String ACCESS_KEY fs.bos.access.key; private static final String SECRET_KEY fs.bos.secret.access.key; private static final String ENDPOINT fs.bos.endpoint; private static final String BUCKET_HIERARCHY fs.bos.bucket.hierarchy;在buildWithReadonlyConfigL60-L69中连接器配置项与 HDFS 配置项的对应关系为BosFile 连接器参数HDFS 客户端配置项access_keyfs.bos.access.keysecret_keyfs.bos.secret.access.keyendpointfs.bos.endpoint—写死fs.bos.bucket.hierarchy falsebucketbos://xxx形式作为 HadoopConf 的 name node key其中fs.bos.bucket.hierarchy被强制设置为false使 BOS 路径以平铺方式寻址bucket参数因此必须带bos://前缀例如bos://my-bucketHadoop 客户端凭此 schema 路由到BaiduBosFileSystem实现。此外模块内还通过 META-INF/services/org.apache.hadoop.fs.FileSystem 注册了该文件系统服务保证客户端类加载时能被 Hadoop SPI 发现。BOS 专属的四个基础参数access_key、secret_key、endpoint、bucket在 BosFileBaseOptions.java 中定义并继承FileBaseSourceOptions复用文件体系通用选项sink 侧的 BosFileSinkOptions 在此之上扩展了写入相关选项。事务写入临时目录 重命名提交is_enable_transaction默认为true官方文档承诺其保证数据在写入目标目录时不丢不重。从源码结构看这一语义由文件连接器基类实现数据先写入tmp_path默认bos://bucket/tmp/seatunnel下的{transactionId}目录任务提交阶段再由聚合 committer 执行“重命名 清理”动作。在 FileSinkAggregatedCommitInfo.java 中可以看到提交逻辑L62-L77for (Map.EntryString, String mvFileEntry : ...) { // first rename temp file hadoopFileSystemProxy.renameFile( mvFileEntry.getKey(), mvFileEntry.getValue(), true); } String transactionDir entry.getKey(); // Data files are already committed after rename; tmp cleanup is ... hadoopFileSystemProxy.deleteFile(transactionDir);即先把tmp_path下的临时文件rename到path目标目录成功后删除事务目录提交失败时则走 BaseFileSinkWriter 中针对未恢复abandoned事务的writeStrategy.abortPrepare(transaction)回滚路径。这也是tmp_path必须是一个 BOS 目录的原因BOS 的重命名语义依赖同一文件系统内的目录移动。当is_enable_transaction true时最终文件名会自动带上${transactionId}_前缀避免重试或并行提交导致的文件名冲突。完整参数列表以下为 官方文档 的完整参数表含源码中确认的默认值参数名类型必填默认值说明pathstringyes-The target directory the sink writes to inside the bucket.桶内写入的目标目录tmp_pathstringno/tmp/seatunnelThe result file will write to a tmp path first and then usemvto submit tmp dir to target dir. Needs a BOS dir.先写临时路径再重命名提交需为 BOS 目录bucketstringyes-The BOS bucket address, for examplebos://my-bucket.access_keystringyes-The Baidu Cloud BOS access key.secret_keystringyes-The Baidu Cloud BOS secret key.endpointstringyes-The BOS endpoint, for examplehttp://bj.bcebos.com.custom_filenamebooleannofalseWhether you need custom the filename.是否自定义文件名file_name_expressionstringno${transactionId}Only used when custom_filename is true.文件名表达式filename_time_formatstringnoyyyy.MM.ddOnly used when custom_filename is true.${now}的时间格式file_format_typestringnocsvFile format type, supported:text,csv,parquet,orc,json,excel,xml,binary,canal_json,debezium_json,maxwell_json.filename_extensionstringno-Override the default file name extensions with custom file name extensions. E.g..xml,.json,dat,.customtypefield_delimiterstringno\001 for text and , for csvOnly used when file_format_type is text and csv.row_delimiterstringno\nOnly used when file_format_type istext,csvandjson.have_partitionbooleannofalseWhether you need processing partitions.是否按字段分区partition_byarrayno-Only used when have_partition is true.partition_dir_expressionstringno${k0}${v0}/${k1}${v1}/.../${kn}${vn}/Only used when have_partition is true.is_partition_field_write_in_filebooleannofalseOnly used when have_partition is true.分区字段是否仍写入文件内容sink_columnsarrayno空When this parameter is empty, all fields are sink columns.限定写入的列留空则写全部列is_enable_transactionbooleannotrueIftrue, data will not be lost or duplicated when written to the target directory. Whentrue,${transactionId}_is automatically prefixed to the file name.batch_sizeintno1000000The maximum number of rows in a file. For SeaTunnel Engine the file row count is jointly decided bybatch_sizeandcheckpoint.interval.compress_codecstringnononeThe compress codec of files. Excel does not support any compression format.xml_root_tagstringnoRECORDSOnly used when file_format is xml.xml_row_tagstringnoRECORDOnly used when file_format is xml.xml_use_attr_formatbooleanno-Only used when file_format is xml.single_file_modebooleannofalseEach parallelism will only output one file. When this parameter is turned on, batch_size will not take effect. The output file name does not have a file block suffix.create_empty_file_when_no_databooleannofalseWhen there is no data synchronization upstream, the corresponding data files are still generated.parquet_avro_write_timestamp_as_int96booleannofalseOnly used when file_format is parquet.parquet_avro_write_fixed_as_int96arrayno-Only used when file_format is parquet.encodingstringnoUTF-8Only used when file_format_type is json,text,csv,xml.common-optionsobjectno-Sink plugin common parameters, please refer to Sink Common Options for details.参数间的条件依赖在 BosFileSinkFactory.optionRule() 中有精确声明例如file_format_type text/csv时才允许field_delimiter、enable_header_writejson格式仅允许row_delimitercustom_filename true时才启用file_name_expression与filename_time_formathave_partition true时才启用partition_by、partition_dir_expression、is_partition_field_write_in_fileencoding仅对 text/json/csv/xml 四种格式生效parquet 格式的parquet_compress、parquet_avro_write_timestamp_as_int96等参数通过compress_codec及各格式条件分支暴露。特性支持矩阵官方文档声明的 Key Featuresmultimodal多模态以二进制文件格式读写任意格式文件视频、图片等任意文件均可同步到目标位置exactly-once默认使用 2PC commit 保证support multiple table write多表写入支持cdc / timer flush不支持。配置示例示例一text 格式 分区 自定义文件名 列裁剪官方文档给出的完整示例同时演示了have_partition、custom_filename与sink_columnsenv { parallelism 1 job.mode BATCH } sink { BosFile { path /sink bucket bos://sink-bucket access_key your-access-key secret_key your-secret-key endpoint http://bj.bcebos.com 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} filename_time_format yyyy.MM.dd sink_columns [name, age] is_enable_transaction true } }解读几个关键点partition_by [age]按 age 字段值划分目录partition_dir_expression ${k0}${v0}使目标路径形如/sink/age25/is_partition_field_write_in_file true表示分区字段仍保留在文件内容中file_name_expression ${transactionId}_${now}中的${now}按filename_time_format格式化配合is_enable_transaction true的自动前缀可确保多次运行、多 checkpoint 提交的文件互不覆盖sink_columns [name, age]仅写出这两列其余上游字段被丢弃。示例二parquet 格式带压缩官方文档的 parquet 最简示例sink { BosFile { path /sink bucket bos://sink-bucket access_key your-access-key secret_key your-secret-key endpoint http://bj.bcebos.com file_format_type parquet is_enable_transaction true } }仓库 e2e 测试中还有一个更贴近生产的写法参考 fake_to_bos_file_parquet.confsink { BosFile { path/sink/parquet bucket bos://seatunnel-test access_key your-access-key secret_key your-secret-key endpoint http://bj.bcebos.com partition_dir_expression ${k0}${v0} is_partition_field_write_in_file true file_name_expression ${transactionId}_${now} file_format_type parquet filename_time_format yyyy.MM.dd is_enable_transaction true compress_codec gzip } }这里通过compress_codec gzip对输出文件启用压缩注意 excel 格式不支持任何压缩。示例三最简 text 写入sink { BosFile { bucket bos://sink-bucket path /warehouse/table/ file_format_type text access_key your-access-key secret_key your-secret-key endpoint http://bj.bcebos.com row_delimiter \n field_delimiter , is_enable_transaction true } }该写法适合批量导出仓库目录path /warehouse/table/指定桶内目录其余全部走默认值tmp_path默认tmp/seatunnel、batch_size默认 100 万行/文件、事务开启。端到端测试与验证BosFile 的 e2e 测试位于 connector-file-bos-e2eBosFileIT.java 覆盖了 excel、text、json、orc、parquet 五种格式的“FakeSource 写入 BOS → 再从 BOS 读回并断言”的完整链路。由于测试需要真实的 BOS 桶与凭证测试类上标注了Disabled其配置文件中的 AK/SK 使用dummy占位见 fake_to_bos_file_parquet.conf。本地验证时替换为真实凭证并移除Disabled即可运行插件工厂层面还有常驻运行的 BosFileFactoryTest 保证配置校验逻辑不回归。部署与版本记录部署清单小结Zeta 引擎为例将bos-hdfs-sdk 1.0.4-community放入${SEATUNNEL_HOME}/lib下载地址与版本要求见 lib/README.md确认 bucket 名称带bos://前缀、endpoint 与桶所在区域匹配如华北 2 为http://bj.bcebos.com使用sink { BosFile { ... } }编写任务配置参照上文三个示例若使用 Spark/Flink 引擎另需保证集群已集成 Hadoop 2.x。Changelog来自 connector-file-bos.mdChangeVersion[Improve][Connector-V2] Add Hive BOSStorage and align BosFile e2e/docs with CosFiledev[Feature][Connector-V2] Add BosFile source and sink for Baidu Object Storagedev如需从 BOS读取数据如将 BOS 中的 parquet/csv 文件同步到数据库可使用配套的 BosFile source 连接器其鉴权参数与本 sink 一致并共享同一套bos-hdfs-sdk运行时依赖。【免费下载链接】seatunnelSeaTunnel is a multimodal, high-performance, distributed, massive data integration tool.项目地址: https://gitcode.com/GitHub_Trending/se/seatunnel创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
Apache SeaTunnel BosFile Sink 深入解析：基于 HDFS SDK 向百度智能云 BOS 对象存储写入数据
Apache SeaTunnel BosFile Sink 深入解析基于 HDFS SDK 向百度智能云 BOS 对象存储写入数据【免费下载链接】seatunnelSeaTunnel is a multimodal, high-performance, distributed, massive data integration tool.项目地址: https://gitcode.com/GitHub_Trending/se/seatunnelBosFile 是 SeaTunnel 提供的对象存储文件写入连接器它通过百度 BOS HDFS SDK 将同步任务的结果写入百度智能云 BOSBaidu Object Storage桶中的指定路径支持 text、csv、parquet、orc、json、excel、xml、binary 等多种文件格式并默认基于 2PC 事务机制保证 exactly-once 语义。读完本文你可以掌握 BosFile sink 的完整参数配置、自定义文件名与分区目录的写法、基于临时目录 重命名的提交原理以及部署bos-hdfs-sdk运行依赖的注意事项。连接器概览引擎支持与运行前提BosFile sink 支持以下计算引擎SeaTunnel ZetaSparkFlink在部署前需要特别注意依赖与 Hadoop 版本前提引自 官方文档若使用Spark / Flink必须确保集群已集成 Hadoop当前经过测试的 Hadoop 版本为2.x若使用SeaTunnel Zeta 引擎Hadoop 相关 jar 已打包在${SEATUNNEL_HOME}/lib下无需额外处理无论哪种引擎都必须将bos-hdfs-sdk 1.0.4-community放入${SEATUNNEL_HOME}/lib。该 jar 并未发布到 Maven CentralSeaTunnel 也不在 pom 中声明它只能作为运行时依赖手工安装。为什么强调 1.0.4 版本下限仓库中专门维护的 SDK 说明文件 connector-file-bos/lib/README.md 给出了明确解释bos-hdfs-sdkisnot published to Maven Central. SeaTunnel does not declare it as a Maven dependency; users must install the jar atruntimeonly.Use1.0.4. SDK 1.0.3 always callsheadBucketduring FileSystem init and ignoresfs.bos.bucket.hierarchyfalse, which fails when the AK/SK lacksHeadBucketpermission.也就是说 1.0.3 及更早版本在 FileSystem 初始化时总会调用headBucket并忽略fs.bos.bucket.hierarchyfalse配置当 AK/SK 缺少HeadBucket权限时任务会直接失败。下载与安装步骤以该文件及 BosFile source 文档 的说明为准。实现原理把 BOS 当作 Hadoop FileSystemBosFile sink 并不是直接使用 BOS 的 REST API 写对象而是复用了 SeaTunnel 文件连接器体系的Hadoop FileSystem 抽象层从而与其他文件系统S3、COS、OSS 等共享同一套写入、切分、事务提交逻辑。核心实现类链路如下BosFileSinkFactory.java插件工厂factoryIdentifier()返回BosFile对应任务配置中sink { BosFile { ... } }的插件名并声明了OptionRule校验规则——path、bucket、access_key、secret_key、endpoint为必填项其余参数按文件类型、custom_filename、have_partition等条件按需生效BosFileSink.javasink 入口继承通用文件写入基类BaseFileSink仅通过覆写initHadoopConf()注入 BOS 专属的 Hadoop 配置BosConf.java将连接器参数翻译为 HDFS 客户端配置。从 BosConf.java 的源码可以看到关键映射private static final String HDFS_IMPL org.apache.hadoop.fs.bos.BaiduBosFileSystem; private static final String SCHEMA bos; private static final String ACCESS_KEY fs.bos.access.key; private static final String SECRET_KEY fs.bos.secret.access.key; private static final String ENDPOINT fs.bos.endpoint; private static final String BUCKET_HIERARCHY fs.bos.bucket.hierarchy;在buildWithReadonlyConfigL60-L69中连接器配置项与 HDFS 配置项的对应关系为BosFile 连接器参数HDFS 客户端配置项access_keyfs.bos.access.keysecret_keyfs.bos.secret.access.keyendpointfs.bos.endpoint—写死fs.bos.bucket.hierarchy falsebucketbos://xxx形式作为 HadoopConf 的 name node key其中fs.bos.bucket.hierarchy被强制设置为false使 BOS 路径以平铺方式寻址bucket参数因此必须带bos://前缀例如bos://my-bucketHadoop 客户端凭此 schema 路由到BaiduBosFileSystem实现。此外模块内还通过 META-INF/services/org.apache.hadoop.fs.FileSystem 注册了该文件系统服务保证客户端类加载时能被 Hadoop SPI 发现。BOS 专属的四个基础参数access_key、secret_key、endpoint、bucket在 BosFileBaseOptions.java 中定义并继承FileBaseSourceOptions复用文件体系通用选项sink 侧的 BosFileSinkOptions 在此之上扩展了写入相关选项。事务写入临时目录 重命名提交is_enable_transaction默认为true官方文档承诺其保证数据在写入目标目录时不丢不重。从源码结构看这一语义由文件连接器基类实现数据先写入tmp_path默认bos://bucket/tmp/seatunnel下的{transactionId}目录任务提交阶段再由聚合 committer 执行“重命名 清理”动作。在 FileSinkAggregatedCommitInfo.java 中可以看到提交逻辑L62-L77for (Map.EntryString, String mvFileEntry : ...) { // first rename temp file hadoopFileSystemProxy.renameFile( mvFileEntry.getKey(), mvFileEntry.getValue(), true); } String transactionDir entry.getKey(); // Data files are already committed after rename; tmp cleanup is ... hadoopFileSystemProxy.deleteFile(transactionDir);即先把tmp_path下的临时文件rename到path目标目录成功后删除事务目录提交失败时则走 BaseFileSinkWriter 中针对未恢复abandoned事务的writeStrategy.abortPrepare(transaction)回滚路径。这也是tmp_path必须是一个 BOS 目录的原因BOS 的重命名语义依赖同一文件系统内的目录移动。当is_enable_transaction true时最终文件名会自动带上${transactionId}_前缀避免重试或并行提交导致的文件名冲突。完整参数列表以下为 官方文档 的完整参数表含源码中确认的默认值参数名类型必填默认值说明pathstringyes-The target directory the sink writes to inside the bucket.桶内写入的目标目录tmp_pathstringno/tmp/seatunnelThe result file will write to a tmp path first and then usemvto submit tmp dir to target dir. Needs a BOS dir.先写临时路径再重命名提交需为 BOS 目录bucketstringyes-The BOS bucket address, for examplebos://my-bucket.access_keystringyes-The Baidu Cloud BOS access key.secret_keystringyes-The Baidu Cloud BOS secret key.endpointstringyes-The BOS endpoint, for examplehttp://bj.bcebos.com.custom_filenamebooleannofalseWhether you need custom the filename.是否自定义文件名file_name_expressionstringno${transactionId}Only used when custom_filename is true.文件名表达式filename_time_formatstringnoyyyy.MM.ddOnly used when custom_filename is true.${now}的时间格式file_format_typestringnocsvFile format type, supported:text,csv,parquet,orc,json,excel,xml,binary,canal_json,debezium_json,maxwell_json.filename_extensionstringno-Override the default file name extensions with custom file name extensions. E.g..xml,.json,dat,.customtypefield_delimiterstringno\001 for text and , for csvOnly used when file_format_type is text and csv.row_delimiterstringno\nOnly used when file_format_type istext,csvandjson.have_partitionbooleannofalseWhether you need processing partitions.是否按字段分区partition_byarrayno-Only used when have_partition is true.partition_dir_expressionstringno${k0}${v0}/${k1}${v1}/.../${kn}${vn}/Only used when have_partition is true.is_partition_field_write_in_filebooleannofalseOnly used when have_partition is true.分区字段是否仍写入文件内容sink_columnsarrayno空When this parameter is empty, all fields are sink columns.限定写入的列留空则写全部列is_enable_transactionbooleannotrueIftrue, data will not be lost or duplicated when written to the target directory. Whentrue,${transactionId}_is automatically prefixed to the file name.batch_sizeintno1000000The maximum number of rows in a file. For SeaTunnel Engine the file row count is jointly decided bybatch_sizeandcheckpoint.interval.compress_codecstringnononeThe compress codec of files. Excel does not support any compression format.xml_root_tagstringnoRECORDSOnly used when file_format is xml.xml_row_tagstringnoRECORDOnly used when file_format is xml.xml_use_attr_formatbooleanno-Only used when file_format is xml.single_file_modebooleannofalseEach parallelism will only output one file. When this parameter is turned on, batch_size will not take effect. The output file name does not have a file block suffix.create_empty_file_when_no_databooleannofalseWhen there is no data synchronization upstream, the corresponding data files are still generated.parquet_avro_write_timestamp_as_int96booleannofalseOnly used when file_format is parquet.parquet_avro_write_fixed_as_int96arrayno-Only used when file_format is parquet.encodingstringnoUTF-8Only used when file_format_type is json,text,csv,xml.common-optionsobjectno-Sink plugin common parameters, please refer to Sink Common Options for details.参数间的条件依赖在 BosFileSinkFactory.optionRule() 中有精确声明例如file_format_type text/csv时才允许field_delimiter、enable_header_writejson格式仅允许row_delimitercustom_filename true时才启用file_name_expression与filename_time_formathave_partition true时才启用partition_by、partition_dir_expression、is_partition_field_write_in_fileencoding仅对 text/json/csv/xml 四种格式生效parquet 格式的parquet_compress、parquet_avro_write_timestamp_as_int96等参数通过compress_codec及各格式条件分支暴露。特性支持矩阵官方文档声明的 Key Featuresmultimodal多模态以二进制文件格式读写任意格式文件视频、图片等任意文件均可同步到目标位置exactly-once默认使用 2PC commit 保证support multiple table write多表写入支持cdc / timer flush不支持。配置示例示例一text 格式 分区 自定义文件名 列裁剪官方文档给出的完整示例同时演示了have_partition、custom_filename与sink_columnsenv { parallelism 1 job.mode BATCH } sink { BosFile { path /sink bucket bos://sink-bucket access_key your-access-key secret_key your-secret-key endpoint http://bj.bcebos.com 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} filename_time_format yyyy.MM.dd sink_columns [name, age] is_enable_transaction true } }解读几个关键点partition_by [age]按 age 字段值划分目录partition_dir_expression ${k0}${v0}使目标路径形如/sink/age25/is_partition_field_write_in_file true表示分区字段仍保留在文件内容中file_name_expression ${transactionId}_${now}中的${now}按filename_time_format格式化配合is_enable_transaction true的自动前缀可确保多次运行、多 checkpoint 提交的文件互不覆盖sink_columns [name, age]仅写出这两列其余上游字段被丢弃。示例二parquet 格式带压缩官方文档的 parquet 最简示例sink { BosFile { path /sink bucket bos://sink-bucket access_key your-access-key secret_key your-secret-key endpoint http://bj.bcebos.com file_format_type parquet is_enable_transaction true } }仓库 e2e 测试中还有一个更贴近生产的写法参考 fake_to_bos_file_parquet.confsink { BosFile { path/sink/parquet bucket bos://seatunnel-test access_key your-access-key secret_key your-secret-key endpoint http://bj.bcebos.com partition_dir_expression ${k0}${v0} is_partition_field_write_in_file true file_name_expression ${transactionId}_${now} file_format_type parquet filename_time_format yyyy.MM.dd is_enable_transaction true compress_codec gzip } }这里通过compress_codec gzip对输出文件启用压缩注意 excel 格式不支持任何压缩。示例三最简 text 写入sink { BosFile { bucket bos://sink-bucket path /warehouse/table/ file_format_type text access_key your-access-key secret_key your-secret-key endpoint http://bj.bcebos.com row_delimiter \n field_delimiter , is_enable_transaction true } }该写法适合批量导出仓库目录path /warehouse/table/指定桶内目录其余全部走默认值tmp_path默认tmp/seatunnel、batch_size默认 100 万行/文件、事务开启。端到端测试与验证BosFile 的 e2e 测试位于 connector-file-bos-e2eBosFileIT.java 覆盖了 excel、text、json、orc、parquet 五种格式的“FakeSource 写入 BOS → 再从 BOS 读回并断言”的完整链路。由于测试需要真实的 BOS 桶与凭证测试类上标注了Disabled其配置文件中的 AK/SK 使用dummy占位见 fake_to_bos_file_parquet.conf。本地验证时替换为真实凭证并移除Disabled即可运行插件工厂层面还有常驻运行的 BosFileFactoryTest 保证配置校验逻辑不回归。部署与版本记录部署清单小结Zeta 引擎为例将bos-hdfs-sdk 1.0.4-community放入${SEATUNNEL_HOME}/lib下载地址与版本要求见 lib/README.md确认 bucket 名称带bos://前缀、endpoint 与桶所在区域匹配如华北 2 为http://bj.bcebos.com使用sink { BosFile { ... } }编写任务配置参照上文三个示例若使用 Spark/Flink 引擎另需保证集群已集成 Hadoop 2.x。Changelog来自 connector-file-bos.mdChangeVersion[Improve][Connector-V2] Add Hive BOSStorage and align BosFile e2e/docs with CosFiledev[Feature][Connector-V2] Add BosFile source and sink for Baidu Object Storagedev如需从 BOS读取数据如将 BOS 中的 parquet/csv 文件同步到数据库可使用配套的 BosFile source 连接器其鉴权参数与本 sink 一致并共享同一套bos-hdfs-sdk运行时依赖。【免费下载链接】seatunnelSeaTunnel is a multimodal, high-performance, distributed, massive data integration tool.项目地址: https://gitcode.com/GitHub_Trending/se/seatunnel创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
