使用JAVA API 解析ORC File

时间:2021-07-30 15:54:56

使用JAVA API 解析ORC File

orc File 的解析过程中,使用FileInputFormat的getSplits(conf, 1)函数,

然后使用 RecordReaderreader = in.getRecordReader(splits[0], conf, Reporter.NULL);

解析ORCfile, 发现当ORC 文件的比较大的时候,超过256M时,不能读取所有的数据。

比如一个ORC 文件有300M,共有180万的条数据,使用上面的方法只能读取出110万的数据,剩下70万的数据读取不出。

使用的读取示例源码如下:

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hive.ql.io.orc.*;
import org.apache.hadoop.hive.ql.io.orc.RecordReader;
import org.apache.hadoop.hive.serde2.SerDeException;
import org.apache.hadoop.hive.serde2.objectinspector.StructField;
import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector;
import org.apache.hadoop.mapred.*;
import org.apache.orc.TypeDescription; import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties; public void readOrcFile(String fileName) throws SerDeException, IOException {
JobConf conf = new JobConf(hadoopConf);
Path orcFilePath = new Path(fileName);
StringBuilder allColumns = new StringBuilder();
StringBuilder allColumnTypes = new StringBuilder();
Properties p = new Properties();
p.setProperty("columns", "url,word,freq,weight");
p.setProperty("columns.types", "string:string:string:string");
OrcSerde serde = new OrcSerde();
serde.initialize(conf, p);
StructObjectInspector inspector = (StructObjectInspector) serde.getObjectInspector();
OrcInputFormat in = new OrcInputFormat();
FileInputFormat.setInputPaths(conf, orcFilePath);
InputSplit[] splits = in.getSplits(conf, 1);
System.out.println("splits.length==" + splits.length);
RecordReader reader = in.getRecordReader(splits[0], conf, Reporter.NULL); Long count = 0 L;
while (reader.next(key, value)) {
count ++;
}
reader.close();
}

org.apache.hadoop.mapred.InputFormat接口的getSplits方法定义如下:

InputSplit[] getSplits(JobConf job,
int numSplits)
throws IOException

其中numSplits参数的含义时期望得到分片数, 如上的例子中,期望输入文件的分片为1个,如果ORC文件有多个分片则会被合并成一个分片。但是hdfs的中设置的一个分片最大为256M,所以合并成1个分片就会少300-256=44M的数据,造成了上面的问题。

如果 numSplits 参数的值设置为小于0的负数,则会按照ORC File的正常的 stripe个数生成split。

InputSplit[] splits = in.getSplits(conf, -1) 得到的 splits 个数是6个,6个splits中记录数是预期

中的180条。

(二)使用 org.apache.hadoop.hive.ql.io.orc.Reader 类读取ORC文件

可以通过reader.getSchema(); // 获取ORC文件的schema文件。

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hive.ql.io.orc.*;
import org.apache.hadoop.hive.ql.io.orc.RecordReader;
import org.apache.hadoop.hive.serde2.SerDeException;
import org.apache.hadoop.hive.serde2.objectinspector.StructField;
import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector;
import org.apache.hadoop.mapred.*;
import org.apache.orc.TypeDescription; import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties; public void readOrc(String INPUT) throws IOException {
Configuration conf = new Configuration();
Path file_in = new Path(INPUT);
Reader reader = OrcFile.createReader(FileSystem.getLocal(conf), file_in); TypeDescription schema = reader.getSchema(); // 获取ORC文件的schema文件
System.out.println(schema.toJson());
System.out.println(schema.toString());
System.out.println("--------------------------------"); StructObjectInspector inspector = (StructObjectInspector) reader.getObjectInspector();
RecordReader records = reader.rows();
Object row = null;
Long count = 0L;
while (records.hasNext()) {
row = records.next(row);
// System.out.println(row.toString());
count++;
List value_lst = inspector.getStructFieldsDataAsList(row);
}
System.out.println("--------total line=" + count);
}