使用Spark saveAsNewAPIHadoopFile将CSV转Avro时遇空指针异常求助

看起来你踩了一个很常见的坑：直接把原始的CSV行数据（Text类型）塞进AvroKey，没有将数据解析成符合你定义的Avro Schema的GenericRecord对象，这才导致了Avro在序列化时抛出NullPointerException。

错误根源分析 #

你用KeyValueTextInputFormat读取CSV得到的是(Text, Text)类型的RDD，但Avro需要的是严格匹配你topLevelRecord Schema的GenericRecord实例。你现在直接把这个元组传给AvroKey，Avro完全无法识别这个结构，自然会报"null of org.srdevin.avro.topLevelRecord"的错误。

另外，KeyValueTextInputFormat默认是按制表符分割每行的键值，这和普通逗号分隔的CSV结构不匹配，所以你应该换用TextInputFormat来读取每行完整的CSV数据。

修正后的完整代码 #

下面是调整后的代码，核心是添加了CSV到GenericRecord的解析逻辑：

import org.apache.avro.generic.GenericData
import org.apache.avro.generic.GenericRecord
import org.apache.avro.mapred.AvroKey
import org.apache.hadoop.io.{LongWritable, NullWritable, Text}
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat
import org.apache.hadoop.mapreduce.Job
import org.apache.avro.mapreduce.AvroJob
import org.apache.avro.mapreduce.AvroKeyOutputFormat
import java.io.File

val in = "test.csv"
val out = "csvToAvroOutput"
val schema = new org.apache.avro.Schema.Parser().parse(new File("/path/to/test.avsc"))

// 改用TextInputFormat读取完整的CSV行，而不是KeyValue分割
val hadoopRDD = sc.newAPIHadoopFile(
  in,
  classOf[TextInputFormat],
  classOf[LongWritable],
  classOf[Text]
)

// 将每行CSV解析为符合Schema的GenericRecord
val avroKeyRDD = hadoopRDD.map { case (_, lineText) =>
  val line = lineText.toString.trim
  if (line.isEmpty) null // 跳过空行
  else {
    // 假设CSV是逗号分隔，若字段包含逗号/引号，建议用opencsv等专业库解析
    val fields = line.split(",")
    val record = new GenericData.Record(schema)
    
    // 按Schema字段名填充数据，空字段会使用Schema定义的default值
    if (fields.length >= 1 && fields(0).nonEmpty) record.put("name", fields(0))
    if (fields.length >= 2 && fields(1).nonEmpty) record.put("age", fields(1))
    
    new AvroKey[GenericRecord](record)
  }
}.filter(_ != null) // 过滤空行生成的null

// 保存为Avro文件
val job = Job.getInstance
AvroJob.setOutputKeySchema(job, schema)

avroKeyRDD.map((_, NullWritable.get()))
  .saveAsNewAPIHadoopFile(
    out,
    classOf[AvroKey[GenericRecord]],
    classOf[NullWritable],
    classOf[AvroKeyOutputFormat[GenericRecord]],
    job.getConfiguration
  )

关键注意事项 #

1. CSV解析的健壮性：如果你的CSV包含带引号的字段、内嵌逗号等复杂情况，不要用split(",")，建议引入opencsv依赖，使用CSVReader来解析，避免数据解析错误。
2. 空值处理：你的Schema已经给字段定义了null默认值，所以未填充的字段会自动使用默认值，无需额外处理。
3. InputFormat选择：TextInputFormat适合读取普通每行一条记录的CSV，而KeyValueTextInputFormat仅适用于键值对格式的文本文件。

内容的提问来源于stack exchange，提问作者RaviNK259