Scala/SparkでRDDからデータを取り出す

Question

だから、私は大きなデータセットを持っています。これは、stackoverflowユーザベースのサンプルです。次のようにこのデータセットから1行がある：

<row Id="42" Reputation="11849" CreationDate="2008-08-01T13:00:11.640" DisplayName="Coincoin" LastAccessDate="2014-01-18T20:32:32.443" WebsiteUrl="" Location="Montreal, Canada" AboutMe="A guy with the attention span of a dead goldfish who has been having a blast in the industry for more than 10 years.&#xD;&#xA;&#xD;&#xA;Mostly specialized in game and graphics programming, from custom software 3D renderers to accelerated hardware pipeline programming." Views="648" UpVotes="337" DownVotes="40" Age="35" AccountId="33" /> 

私はそれが「35」私は希望であることが、この例では、「11849」で、年齢から数この場合には、評判から番号を抽出したいと思いますそれらをフロートとして持つのが好きです。ファイルはHDFSに位置しています

ので、それは私が引用して、それを分割するときに評判がインデックス23でインデックス3と年齢であるマークが、どのように私はこれらを割り当てない

val linesWithAge = lines.filter(line => line.contains("Age=")) //This is filtering data which doesnt have age 
    val repSplit = linesWithAge.flatMap(line => line.split("\"")) //Here I am trying to split the data where there is a " 

RDD

形式で提供されます私は浮動小数点としてそれらを使用することができますマップまたは変数に。 また、RDDのすべての回線でこれを行う必要があります。

EDIT：

val linesWithAge = lines.filter(line => line.contains("Age=")) //transformations from the original input data 
    val repSplit = linesWithAge.flatMap(line => line.split("\"")) 
    val withIndex = repSplit.zipWithIndex 
    val indexKey = withIndex.map{case (k,v) => (v,k)} 
    val b = indexKey.lookup(3) 
    println(b) 

だから、配列にインデックスを追加して、今、私は変数に代入することに成功し管理しているが、私は唯一のRDDに一つの項目にそれを行うことができ、誰もが知っていなければ私はどのようにすべての項目にそれを行うことができますか？

Answer 1

まず、あなたがして、あなたのライン（getValueForKeyAs[T]）の指定されたキーの値を抽出機能がない必要があります。

val rdd = linesWithAge.map(line => (getValueForKeyAs[Float](line,"Age"), getValueForKeyAs[Float](line,"Reputation"))) 

これはあなたのタイプRDD[(Float,Float)]

getValueForKeyAsのRDDを与えるべきである可能性があり

def getValueForKeyAs[A](line:String, key:String) : A = { 
    val res = line.split(key+"=") 
    if(res.size==1) throw new RuntimeException(s"no value for key $key") 
    val value = res(1).split("\"")(1) 
    return value.asInstanceOf[A] 
} 

Answer 2

私たちがやりたいことは、元のデータセットの各要素を変換することです（ RDD）を数値として(Reputation, Age)を含むタプルに変換する。

一つの可能​​なアプローチは、このように、要素「年齢」と「評判」の値を抽出するために、文字列操作を使用して、RDDの各要素を変換することである。

// define a function to extract the value of an element, given the name 
def findElement(src: Array[String], name:String):Option[String] = { 
    for { 
    entry <- src.find(_.startsWith(name)) 
    value <- entry.split("\"").lift(1) 
    } yield value 
} 

我々は、その機能を使用します

val reputationByAge = lines.flatMap{line => 
    val elements = line.split(" ") 
    for { 
     age <- findElement(elements, "Age") 
     rep <- findElement(elements, "Reputation") 
    } yield (rep.toInt, age.toInt) 
} 

これを行う前に、「年齢」をフィルタリングする必要はありません。 「年齢」または「評判」がないレコードを処理する場合、findElementはNoneを返します。以後、for-comprehensionの結果はNoneとなり、の操作ではが平文になります。

この問題にアプローチするより良い方法は、構造化されたXMLデータを処理していることを認識することです。 ScalaはXMLのためのビルトインサポートを提供しますので、我々はこれを行うことができます：

import scala.xml.XML 
import scala.xml.XML._ 

// help function to map Strings to Option where empty strings become None 
def emptyStrToNone(str:String):Option[String] = if (str.isEmpty) None else Some(str) 

val xmlReputationByAge = lines.flatMap{line => 
    val record = XML.loadString(line) 
    for {   
     rep <- emptyStrToNone((record \ "@Reputation").text) 
     age <- emptyStrToNone((record \ "@Age").text) 
    } yield (rep.toInt, age.toInt) 
} 

この方法は、右の属性を抽出するためにXMLレコードの構造に依存しています。前述のように、私たちは必要な情報がすべて含まれていないレコードを削除するには、Optionの値とflatMapの組み合わせを使用します。