本文介紹了java 讀寫(xiě)Parquet格式的數(shù)據(jù),分享給大家,具體如下:
創(chuàng)新互聯(lián)公司堅(jiān)持“要么做到,要么別承諾”的工作理念,服務(wù)領(lǐng)域包括:成都網(wǎng)站建設(shè)、成都做網(wǎng)站、企業(yè)官網(wǎng)、英文網(wǎng)站、手機(jī)端網(wǎng)站、網(wǎng)站推廣等服務(wù),滿足客戶于互聯(lián)網(wǎng)時(shí)代的元江縣網(wǎng)站設(shè)計(jì)、移動(dòng)媒體設(shè)計(jì)的需求,幫助企業(yè)找到有效的互聯(lián)網(wǎng)解決方案。努力成為您成熟可靠的網(wǎng)絡(luò)建設(shè)合作伙伴!
import java.io.BufferedReader; import java.io.File; import java.io.FileReader; import java.io.IOException; import java.util.Random; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; import org.apache.log4j.Logger; import org.apache.parquet.example.data.Group; import org.apache.parquet.example.data.GroupFactory; import org.apache.parquet.example.data.simple.SimpleGroupFactory; import org.apache.parquet.hadoop.ParquetReader; import org.apache.parquet.hadoop.ParquetReader.Builder; import org.apache.parquet.hadoop.ParquetWriter; import org.apache.parquet.hadoop.example.GroupReadSupport; import org.apache.parquet.hadoop.example.GroupWriteSupport; import org.apache.parquet.schema.MessageType; import org.apache.parquet.schema.MessageTypeParser; public class ReadParquet { static Logger logger=Logger.getLogger(ReadParquet.class); public static void main(String[] args) throws Exception { // parquetWriter("test\\parquet-out2","input.txt"); parquetReaderV2("test\\parquet-out2"); } static void parquetReaderV2(String inPath) throws Exception{ GroupReadSupport readSupport = new GroupReadSupport(); Builder<Group> reader= ParquetReader.builder(readSupport, new Path(inPath)); ParquetReader<Group> build=reader.build(); Group line=null; while((line=build.read())!=null){ Group time= line.getGroup("time", 0); //通過(guò)下標(biāo)和字段名稱(chēng)都可以獲取 /*System.out.println(line.getString(0, 0)+"\t"+ line.getString(1, 0)+"\t"+ time.getInteger(0, 0)+"\t"+ time.getString(1, 0)+"\t");*/ System.out.println(line.getString("city", 0)+"\t"+ line.getString("ip", 0)+"\t"+ time.getInteger("ttl", 0)+"\t"+ time.getString("ttl2", 0)+"\t"); //System.out.println(line.toString()); } System.out.println("讀取結(jié)束"); } //新版本中new ParquetReader()所有構(gòu)造方法好像都棄用了,用上面的builder去構(gòu)造對(duì)象 static void parquetReader(String inPath) throws Exception{ GroupReadSupport readSupport = new GroupReadSupport(); ParquetReader<Group> reader = new ParquetReader<Group>(new Path(inPath),readSupport); Group line=null; while((line=reader.read())!=null){ System.out.println(line.toString()); } System.out.println("讀取結(jié)束"); } /** * * @param outPath 輸出Parquet格式 * @param inPath 輸入普通文本文件 * @throws IOException */ static void parquetWriter(String outPath,String inPath) throws IOException{ MessageType schema = MessageTypeParser.parseMessageType("message Pair {\n" + " required binary city (UTF8);\n" + " required binary ip (UTF8);\n" + " repeated group time {\n"+ " required int32 ttl;\n"+ " required binary ttl2;\n"+ "}\n"+ "}"); GroupFactory factory = new SimpleGroupFactory(schema); Path path = new Path(outPath); Configuration configuration = new Configuration(); GroupWriteSupport writeSupport = new GroupWriteSupport(); writeSupport.setSchema(schema,configuration); ParquetWriter<Group> writer = new ParquetWriter<Group>(path,configuration,writeSupport); //把本地文件讀取進(jìn)去,用來(lái)生成parquet格式文件 BufferedReader br =new BufferedReader(new FileReader(new File(inPath))); String line=""; Random r=new Random(); while((line=br.readLine())!=null){ String[] strs=line.split("\\s+"); if(strs.length==2) { Group group = factory.newGroup() .append("city",strs[0]) .append("ip",strs[1]); Group tmpG =group.addGroup("time"); tmpG.append("ttl", r.nextInt(9)+1); tmpG.append("ttl2", r.nextInt(9)+"_a"); writer.write(group); } } System.out.println("write end"); writer.close(); } }
說(shuō)下schema(寫(xiě)Parquet格式數(shù)據(jù)需要schema,讀取的話"自動(dòng)識(shí)別"了schema)
/* * 每一個(gè)字段有三個(gè)屬性:重復(fù)數(shù)、數(shù)據(jù)類(lèi)型和字段名,重復(fù)數(shù)可以是以下三種: * required(出現(xiàn)1次) * repeated(出現(xiàn)0次或多次) * optional(出現(xiàn)0次或1次) * 每一個(gè)字段的數(shù)據(jù)類(lèi)型可以分成兩種: * group(復(fù)雜類(lèi)型) * primitive(基本類(lèi)型) * 數(shù)據(jù)類(lèi)型有 * INT64, INT32, BOOLEAN, BINARY, FLOAT, DOUBLE, INT96, FIXED_LEN_BYTE_ARRAY */
這個(gè)repeated和required 不光是次數(shù)上的區(qū)別,序列化后生成的數(shù)據(jù)類(lèi)型也不同,比如repeqted修飾 ttl2 打印出來(lái)為 WrappedArray([7,7_a]) 而 required修飾 ttl2 打印出來(lái)為 [7,7_a] 除了用MessageTypeParser.parseMessageType類(lèi)生成MessageType 還可以用下面方法
(注意這里有個(gè)坑--spark里會(huì)有這個(gè)問(wèn)題--ttl2這里 as(OriginalType.UTF8) 和 required binary city (UTF8)作用一樣,加上UTF8,在讀取的時(shí)候可以轉(zhuǎn)為StringType,不加的話會(huì)報(bào)錯(cuò) [B cannot be cast to java.lang.String )
/*MessageType schema = MessageTypeParser.parseMessageType("message Pair {\n" + " required binary city (UTF8);\n" + " required binary ip (UTF8);\n" + "repeated group time {\n"+ "required int32 ttl;\n"+ "required binary ttl2;\n"+ "}\n"+ "}");*/ //import org.apache.parquet.schema.Types; MessageType schema = Types.buildMessage() .required(PrimitiveTypeName.BINARY).as(OriginalType.UTF8).named("city") .required(PrimitiveTypeName.BINARY).as(OriginalType.UTF8).named("ip") .repeatedGroup().required(PrimitiveTypeName.INT32).named("ttl") .required(PrimitiveTypeName.BINARY).as(OriginalType.UTF8).named("ttl2") .named("time") .named("Pair");
解決 [B cannot be cast to java.lang.String 異常:
1.要么生成parquet文件的時(shí)候加個(gè)UTF8
2.要么讀取的時(shí)候再提供一個(gè)同樣的schema類(lèi)指定該字段類(lèi)型,比如下面:
maven依賴(lài)(我用的1.7)
<dependency> <groupId>org.apache.parquet</groupId> <artifactId>parquet-hadoop</artifactId> <version>1.7.0</version> </dependency>
以上就是本文的全部?jī)?nèi)容,希望對(duì)大家的學(xué)習(xí)有所幫助,也希望大家多多支持創(chuàng)新互聯(lián)。
新聞名稱(chēng):java讀寫(xiě)Parquet格式的數(shù)據(jù)的示例代碼
網(wǎng)站URL:http://jinyejixie.com/article28/gpspcp.html
成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián),為您提供云服務(wù)器、移動(dòng)網(wǎng)站建設(shè)、Google、定制網(wǎng)站、網(wǎng)頁(yè)設(shè)計(jì)公司、面包屑導(dǎo)航
聲明:本網(wǎng)站發(fā)布的內(nèi)容(圖片、視頻和文字)以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主,如果涉及侵權(quán)請(qǐng)盡快告知,我們將會(huì)在第一時(shí)間刪除。文章觀點(diǎn)不代表本網(wǎng)站立場(chǎng),如需處理請(qǐng)聯(lián)系客服。電話:028-86922220;郵箱:631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載,或轉(zhuǎn)載時(shí)需注明來(lái)源: 創(chuàng)新互聯(lián)