alibaba / alibaba/DataX

支持写RCfile文件

Open
#1,968 0 comments 0 reactions 0 assignees View on GitHub
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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.