spark中的自定义输入阅读器答案

【问题标题】：Custom input reader in sparkspark中的自定义输入阅读器
【发布时间】：2023-03-31 13:39:01
【问题描述】：

我是 Spark 的新手，想将页面记录从 Wikipedia 转储加载到 RDD。

我尝试使用 hadoop 流中提供的记录阅读器，但不知道如何使用它。谁能帮我用下面的代码创建一个不错的带有页面记录的 RDD？

import org.apache.hadoop.io.Text
import org.apache.hadoop.streaming.StreamXmlRecordReader

import org.apache.hadoop.mapred.JobConf
import org.apache.spark.SparkConf
import org.apache.spark.SparkContext

object WikiTest {

  def main(args: Array[String]) {

    // configuration
    val sparkConf = new SparkConf()
      .setMaster("local[4]")
      .setAppName("WikiDumpTest")

    val jobConf = new JobConf()
    jobConf.set("input", "enwikisource-20140906-pages-articles-multistream.xml")
    jobConf.set("stream.recordreader.class", "org.apache.hadoop.streaming.StreamXmlRecordReader")
    jobConf.set("stream.recordreader.begin", "<page>")
    jobConf.set("stream.recordreader.end", "</page>")
      
    val sparkContext = new SparkContext(sparkConf)
  
    // read data
    val wikiData = sparkContext.hadoopRDD(
        jobConf, 
        classOf[StreamXmlRecordReader],
        classOf[Text], 
        classOf[Text])

    // count rows
    println(wikiData.count)
  }
}

Spark 似乎拒绝使用 StreamXmlRecordReader。我收到以下错误：

[错误] 发现：类[org.apache.hadoop.streaming.StreamXmlRecordReader (classOf[org.apache.hadoop.streaming.StreamXmlRecordReader])

[error] required: Class[_ <: org.apache.hadoop.mapreduce.inputformat>
[错误] classOf[StreamXmlRecordReader]

如果我忽略 Eclispse 的警告并启动 prgramm，我会遇到 java.lang.ClassNotFoundException。

【问题讨论】：

标签： scala hadoop apache-spark hadoop-streaming

【解决方案1】：

您应该使用classOf[org.apache.hadoop.streaming.StreamInputFormat] 而不是classOf[StreamXmlRecordReader]。

java.lang.ClassNotFoundException 是因为你想运行你的类WikiTest 但它不存在，因为它无法编译。

【讨论】：

【解决方案2】：

您拥有java.lang.ClassNotFoundException 是因为您试图在 spark (StreamXmlRecordReader) 中使用外部依赖项。您必须创建一个 fat jar，并将其部署在 Spark 中。

这是一个关于如何创建此类 jar 的好示例：gradle fat jar tutorial

如果您在解析 XML 文件时遇到问题，也可以查看这里： parsing tutorial

【讨论】：