如何从 HDFS 检索 Avro 数据？答案

【问题标题】：How to retrieve Avro data from HDFS?如何从 HDFS 检索 Avro 数据？
【发布时间】：2015-02-18 02:35:41
【问题描述】：

我已经为它创建了一个 JSON 数据和一个 Avro 架构：

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

和

{ “类型”：“记录”，“名称”：“twitter_schema”，“命名空间”：“com.miguno.avro”，“字段”：[{ “名称”：“用户名”， “类型”：“字符串”， "doc" : "Twitter.com 上用户帐户的名称" }, { “名称”：“推文”， “类型”：“字符串”， "doc" : "用户推特消息的内容" }, { “名称”：“时间戳”， “类型”：“长”， "doc" : "Unix 纪元时间，以秒为单位" } ], "doc:" : "用于存储 Twitter 消息的基本模式" }

然后我将其转换为 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.avro.mapred.AvroWrapper 在 org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:1185) 在 org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1174) 在 org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1173) 在 scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59) 在 scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:47)

有什么解决办法？

【问题讨论】：

标签： scala apache-spark avro

【解决方案1】：

Spark 正在尝试 ser/de 您的 avro 数据，但它不是“java 可序列化”（默认 ser.spark 中使用）。

你有几个选择：

从包装器中提取通用记录并将每条记录映射到某个可序列化的结构
生成特定的记录类并使用它们而不是通用记录（您仍然需要从包装器中提取记录）
启用 kryo 序列化（这将在某些情况下工作）

请注意，记录在内部被重用，因此如果您执行例如 rdd.collect ，您最终将得到所有记录具有相同值。在进行收集之前将原始输入数据映射到其他内容可以解决您在进行复制时遇到的问题。

【讨论】：