01
描述
${SEATUNNEL_HOME}/lib 目录下确认这个 JAR 包是否存在。02
主要特性
03
选项
| 选项 | 类型 | 必需 | 默认值 | 备注 |
s3n://seatunnel-test，如果您使用的是 s3a 协议，此参数应为 s3a://seatunnel-test。org.apache.hadoop.fs.s3a.SimpleAWSCredentialsProvider 和 com.amazonaws.auth.InstanceProfileCredentialsProvider。hadoop_s3_properties {
"fs.s3a.buffer.dir" = "/data/st_test/s3a"
"fs.s3a.fast.upload.buffer" = "disk"
}
custom_filename 为 true 时使用file_name_expression 描述了将创建到 path 中的文件表达式。我们可以在 file_name_expression 中添加变量 ${now} 或 ${uuid}，例如 test_${uuid}_${now}，${now} 代表当前时间，其格式可以通过指定选项 filename_time_format 来定义。is_enable_transaction 为 true，我们将在文件名的开头自动添加${transactionId}_。custom_filename 为 true 时使用file_name_expression 参数中的格式为 xxxx-${now} 时，filename_time_format 可以指定路径的时间格式，默认值为 yyyy.MM.dd。常用的时间格式列于下表中：| 符号 | 描述 |
txt。file_format 为 text 时需要。file_format 为 text 时需要。have_partition 为 true 时使用。have_partition 为 true 时使用。partition_by，我们将根据分区信息生成相应的分区目录，并将最终文件放在分区目录中。partition_dir_expression 是 ${k0}=${v0}/${k1}=${v1}/.../${kn}=${vn}/。k0 是第一个分区字段，v0 是第一个分区字段的值。have_partition 为 true 时使用。is_partition_field_write_in_file 为 true，分区字段及其值将写入数据文件中。false。is_enable_transaction 为 true，我们将确保在写入目标目录时数据不会丢失或重复。is_enable_transaction 为 true，我们将在文件头部自动添加 ${transactionId}_。true。batch_size 和 checkpoint.interval 共同决定。如果 checkpoint.interval 的值足够大，当文件中的行数大于 batch_size 时，写入器将写入文件。如果 checkpoint.interval较小，则在新的检查点触发时，写入器将创建一个新文件。lzo nonelzo nonelzo nonelzo snappy lz4 zlib nonelzo snappy lz4 gzip brotli zstd none044
示例
have_partition、custom_filename、sink_columns 和 com.amazonaws.auth.InstanceProfileCredentialsProvider 的配置示例：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"
}
}
org.apache.hadoop.fs.s3a.SimpleAWSCredentialsProvider进行配置：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"
}
}
org.apache.hadoop.fs.s3a.SimpleAWSCredentialsProvider：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"
}
05
更新日志
06
下一版本