如何从HDFS中检索Avro数据？

Question

{"username":"miguno","tweet":"Rock: Nerf paper, scissors is fine.","timestamp": 1366150681 }
{"username":"BlizzardCS","tweet":"Works as intended.  Terran is IMBA.","timestamp": 1366154481 }

{“type”：“record”、“name”：“twitter_schema”、“namespace”：“com.miguno.avro”、“fields”:[{“name”：“username”、“type”：“string”、“doc”：“Twitter.com上的用户帐户名称”}、{“name”：“tweet”、“type”：“string”、“doc”：“用户的Twitter消息内容”}、{“name”：“timestamp”、“type”：“long”、“doc”：“Unix epoch time in seconds”}]、“

然后我将其转换为Avro，如下所示：

java -jar ~/avro-tools-1.7.4.jar fromjson --schema-file twitter.avsc twitter.json > twitter.avro

将该文件放在hdfs上，下面是：

hadoop fs -copyFromLocal twitter.avro <path>

然后我在Spark CLI中发布了代码：

import org.apache.avro.generic.GenericRecord
import org.apache.avro.mapred.{AvroInputFormat, AvroWrapper}
import org.apache.hadoop.io.NullWritable

val path = "hdfs:///path/to/your/avro/folder"
val avroRDD = sc.hadoopFile[AvroWrapper[GenericRecord], NullWritable, AvroInputFormat[GenericRecord]](path)

但是，在执行以下操作时：

avroRDD.first

我面临以下例外：

org.apache.spark.sparkException：由于阶段失败而中止作业：阶段7.0（TID 13）中的任务2.0有一个不可序列化的结果:org.apache.spark.scheduler.dagscheduler.org$apache$spark$scheduler$dagscheduler$$failjobandindependentstages(Dagscheduler.scala:1185)在org.apache.spark.scheduler.dagscheduler$$anonfun$abortstage$1.应用(dagscheduler.scala:1174)在3）在scala.collection.mutable.resizablearray$class.foreach（resizablearray.scala:59)在scala.collection.mutable.arrayBuffer.foreach（arrayBuffer.scala:47)

有什么解决办法？

Answer 1

Spark试图ser/de您的avro数据，但它不是“Java可序列化的”（Spark中使用的默认ser.)。

你有几个选择：

• 从包装器中提取泛型记录，并将每个记录映射到某个可序列化结构
• 生成特定的记录类并使用它们而不是一般记录（您仍然需要从包装器中提取记录）
• 启用kryo序列化（这只在某些情况下有效）