支持写RCfile文件
- Dominant language
- Java
- Stars
- 17.4k
- Forks
- 5.7k
- PR merge metrics
- No merged PRs in 30d
Description
最近遇到一个poc,是前五的央企,用的rcfile文件格式,hdfshelper:
``` java
/**
* 写rcfile类型文件
*
* @param lineReceiver
* @param config
* @param fileName
* @param taskPluginCollector
*/
public void rcFileStartWrite(RecordReceiver lineReceiver, Configuration config, String fileName,
TaskPluginCollector taskPluginCollector) {
List columns = config.getListConfiguration(Key.COLUMN);
String compress = config.getString(Key.COMPRESS, null);
List columnNames = getColumnNames(columns);
List columnTypeInspectors = getColumnTypeInspectors(columns);
StructObjectInspector inspector = ObjectInspectorFactory
.getStandardStructObjectInspector(columnNames, columnTypeInspectors);
try {
ColumnarSerDe rcSerde = new ColumnarSerDe();
String types = columnTypeInspectors.stream().map(it -> it.getTypeName()).collect(Collectors.joining(","));
Properties tableProperties = new Properties();
LOG.info("=======" + StringUtils.join(columnNames, ","));
LOG.info("=======" + types);
tableProperties.setProperty("columns", StringUtils.join(columnNames, ","));
tableProperties.setProperty("columns.types", types);
rcSerde.initialize(conf, tableProperties);
RCFileOutputFormat.setColumnNumber(conf, columnNames.size());
FileOutputFormat outFormat = new RCFileOutputFormat();
if (!"NONE".equalsIgnoreCase(compress) && null != compress) {
Class codecClass = getCompressCodec(compress);
if (null != codecClass) {
outFormat.setOutputCompressorClass(conf, codecClass);
}
}
RecordWriter writer = outFormat.getRecordWriter(fileSystem, conf, fileName, Reporter.NULL);
Record record = null;
while ((record = lineReceiver.getFromReader()) != null) {
MutablePair, Boolean> transportResult = transportOneRecord(record, columns, taskPluginCollector);
if (!transportResult.getRight()) {
writer.write(NullWritable.get(), rcSerde.serialize(transportResult.getLeft(), inspector));
}
}
writer.close(Reporter.NULL);
} catch (Exception e) {
String message = String.format("写文件文件[%s]时发生IO异常,请检查您的网络是否正常!", fileName);
LOG.error(message);
Path path = new Path(fileName);
deleteDir(path.getParent());
throw DataXException.asDataXException(HdfsWriterErrorCode.Write_FILE_IO_ERROR, e);
}
}
```
Contributor guide
No contributing guide indexed for this repository
Research direction
Start at the HDFS helper entry point and the rcFileStartWrite example in the issue; trace how existing output formats and compression settings are selected. Verify the RCFile writing path with representative records and configured columns, and confirm that the requested RCFile output is produced successfully.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- hadoop, java
- Domain
- data-engineering
- Issue type
- Feature
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Stale
- Clarity
- Mostly clear
- Newbie friendliness
- 35/100