scala - spark 中的自定义输入阅读器

Question

我是 Spark 的新手，想将页面记录从维基百科转储加载到 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。我收到以下错误:

[error] found : Class[org.apache.hadoop.streaming.StreamXmlRecordReader (classOf[org.apache.hadoop.streaming.StreamXmlRecordReader])

[error] required: Class[_ <: org.apache.hadoop.mapreduce.InputFormat[?,?]]

[error] classOf[StreamXmlRecordReader]

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

Answer 1

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

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