我正在从Cloudera包裹运行带有Spark 0.9.0的CDH 4.4.
我有一堆通过猪的AvroStorage UDF创建的Avro文件.我想在Spark中加载这些文件,使用通用记录或Avro文件上的架构.到目前为止我已经尝试过:
import org.apache.avro.mapred.AvroKey import org.apache.avro.mapreduce.AvroKeyInputFormat import org.apache.hadoop.io.NullWritable import org.apache.commons.lang.StringEscapeUtils.escapecsv import org.apache.hadoop.fs.Path import org.apache.hadoop.fs.FileSystem import org.apache.hadoop.conf.Configuration import java.net.URI import java.io.BufferedInputStream import java.io.File import org.apache.avro.generic.{GenericDatumReader,GenericRecord} import org.apache.avro.specific.SpecificDatumReader import org.apache.avro.file.DataFileStream import org.apache.avro.io.DatumReader import org.apache.avro.file.DataFileReader import org.apache.avro.mapred.FsInput val input = "hdfs://hivecluster2/securityx/web_proxy_mef/2014/05/29/22/part-m-00016.avro" val inURI = new URI(input) val inPath = new Path(inURI) val fsInput = new FsInput(inPath,sc.hadoopConfiguration) val reader = new GenericDatumReader[GenericRecord] val dataFileReader = DataFileReader.openReader(fsInput,reader) val schemaString = dataFileReader.getSchema val buf = scala.collection.mutable.ListBuffer.empty[GenericRecord] while(dataFileReader.hasNext) { buf += dataFileReader.next } sc.parallelize(buf)
解决方法
回答我自己的问题:
import org.apache.spark.SparkContext import org.apache.spark.SparkContext._ import org.apache.avro.generic.GenericRecord import org.apache.avro.mapred.AvroKey import org.apache.avro.mapred.AvroInputFormat import org.apache.avro.mapreduce.AvroKeyInputFormat import org.apache.hadoop.io.NullWritable import org.apache.commons.lang.StringEscapeUtils.escapecsv import org.apache.hadoop.fs.FileSystem import org.apache.hadoop.fs.Path import org.apache.hadoop.conf.Configuration import java.io.BufferedInputStream import org.apache.avro.file.DataFileStream import org.apache.avro.io.DatumReader import org.apache.avro.file.DataFileReader import org.apache.avro.file.DataFileReader import org.apache.avro.generic.{GenericDatumReader,GenericRecord} import org.apache.avro.mapred.FsInput import org.apache.avro.Schema import org.apache.avro.Schema.Parser import org.apache.hadoop.mapred.JobConf import java.io.File import java.net.URI // spark-shell -usejavacp -classpath "*.jar" val input = "hdfs://hivecluster2/securityx/web_proxy_mef/2014/05/29/22/part-m-00016.avro" val jobConf= new JobConf(sc.hadoopConfiguration) val rdd = sc.hadoopFile( input,classOf[org.apache.avro.mapred.AvroInputFormat[GenericRecord]],classOf[org.apache.avro.mapred.AvroWrapper[GenericRecord]],classOf[org.apache.hadoop.io.NullWritable],10) val f1 = rdd.first val a = f1._1.datum a.get("rawLog") // Access avro fields
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。