github/DataX
5
0
Fork 0
mirror of https://github.com/alibaba/DataX.git synced 2025-05-01 18:40:44 +08:00
Code Issues Packages Projects Releases Wiki Activity

gen default parquet schema if config not set

Browse Source
This commit is contained in:
liaoguangwen 2024-10-26 01:26:47 +08:00
parent 3614c2633e
commit 7bb69f6c85
2 changed files with 14 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
hdfswriter/src/main/java/com/alibaba/datax/plugin/writer/hdfswriter/HdfsHelper.java
View File

@ -630,7 +630,14 @@ public class HdfsHelper {
MessageType messageType = null;
ParquetFileProccessor proccessor = null;
Path outputPath = new Path(fileName);
String schema = config.getString(Key.PARQUET_SCHEMA);
String schema = config.getString(Key.PARQUET_SCHEMA, null);
if (schema == null) {
List<Configuration> columns = config.getListConfiguration(Key.COLUMN);
if (columns == null || columns.isEmpty()) {
throw DataXException.asDataXException("parquetSchema or column can't be blank!");
}
schema = HdfsHelper.generateParquetSchemaFromColumnAndType(columns);
}
try {
messageType = MessageTypeParser.parseMessageType(schema);
} catch (Exception e) {

6
hdfswriter/src/main/java/com/alibaba/datax/plugin/writer/hdfswriter/HdfsWriter.java
View File

@ -228,6 +228,12 @@ public class HdfsWriter extends Writer {
String endFullFileName = null;
fileSuffix = UUID.randomUUID().toString().replace('-', '_');
if (fileType.equalsIgnoreCase("PARQUET")) {
if (StringUtils.isNotBlank(this.compress)) {
fileSuffix += "." + this.compress.toLowerCase();
}
fileSuffix += ".parquet";
}
fullFileName = String.format("%s%s%s__%s", defaultFS, storePath, filePrefix, fileSuffix);
endFullFileName = String.format("%s%s%s__%s", defaultFS, endStorePath, filePrefix, fileSuffix);