@

点击蓝字

关注我们

S3File 是一个用于管理 Amazon S3（Simple Storage Service）的 Python 模块。当前，Apache SeaTunnel 已经支持 S3File Sink Connector，为了更好地使用这个 Connector，有必要看一下这篇使用文档指南。

✅文本 (text)
✅CSV
✅Parquet
✅ORC
✅JSON
✅Excel

path [string]

bucket [string]

fs.s3a.endpoint [string]

fs.s3a.aws.credentials.provider [string]

access_key [string]

access_secret [string]

hadoop_s3_properties [map]

hadoop_s3_properties {
      "fs.s3a.buffer.dir" = "/data/st_test/s3a"
      "fs.s3a.fast.upload.buffer" = "disk"
   }

custom_filename [boolean]

file_name_expression [string]

filename_time_format [string]

file_format_type [string]

• 文本 (text)
• JSON
• CSV
• ORC
• Parquet
• Excel

field_delimiter [string]

row_delimiter [string]

have_partition [boolean]

partition_by [array]

partition_dir_expression [string]

is_partition_field_write_in_file [boolean]

sink_columns [array]

is_enable_transaction [boolean]

batch_size [int]

compress_codec [string]

• txt: lzo
 none
  
• JSON: lzo
 none
  
• CSV: lzo
 none
  
• ORC: lzo
 snappy
 lz4
 zlib
 none
  
• Parquet: lzo
 snappy
 lz4
 gzip
 brotli
 zstd
 none
  

常见选项

max_rows_in_memory [int]

sheet_name [string]

  S3File {
    bucket = "s3a://seatunnel-test"
    tmp_path = "/tmp/seatunnel"
    path="/seatunnel/text"
    fs.s3a.endpoint="s3.cn-north-1.amazonaws.com.cn"
    fs.s3a.aws.credentials.provider="com.amazonaws.auth.InstanceProfileCredentialsProvider"
    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
    hadoop_s3_properties {
      "fs.s3a.buffer.dir" = "/data/st_test/s3a"
      "fs.s3a.fast.upload.buffer" = "disk"
    }
  }

  S3File {
    bucket = "s3a://seatunnel-test"
    tmp_path = "/tmp/seatunnel"
    path="/seatunnel/parquet"
    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"
    file_format_type = "parquet"
    hadoop_s3_properties {
      "fs.s3a.buffer.dir" = "/data/st_test/s3a"
      "fs.s3a.fast.upload.buffer" = "disk"
    }
  }

  S3File {
    bucket = "s3a://seatunnel-test"
    tmp_path = "/tmp/seatunnel"
    path="/seatunnel/orc"
    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"
    file_format_type = "orc"
  }

2.3.0-beta 2022-10-20

• 添加 S3File Sink 连接器

2.3.0 2022-12-30

• Bug修复
• 当上游字段为空时会抛出 NullPointerException
• Sink 列映射失败
• 从状态中恢复写入器时直接获取事务失败 (3258)
• 修复了以下导致数据写入文件失败的错误:
• 功能
• 允许用户添加额外的 Hadoop-S3 参数
• 允许使用 S3A 协议
• 解耦 Hadoop-AWS 依赖
• 支持 S3A 协议 (3632)
• 支持设置每个文件的批处理大小 (3625)
• 设置 S3 AK 为可选项 (3688)

• [优化]支持文件压缩（3699）

一键三连-点赞在看转发⭐️！