Scala/Spark에서 RDD로부터 데이터 추출하기

Question

그래서 stackoverflow 사용자베이스의 샘플 인 큰 데이터 세트가 있습니다. 다음과 같이 데이터 세트에서 한 줄은 다음과 같습니다

<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에 위치한

그래서 내가 인용으로 분할 할 때 RDD가

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 " 

그래서 명성 지수 23 인덱스 세 나이에 표시하는 형식으로 제공하지만 이러한 방법을 지정 않는다 지도 나 변수를 플로트로 사용할 수 있습니다. 또한 RDD의 모든 회선에서이 작업을 수행해야합니다.

편집 : 배열에 인덱스를 추가하고 지금은 성공적으로 나는 RDD에서 하나 개의 항목에 그것을 할 만 할 수있는 변수에 할당 관리하지만 한 경우

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) 

그래서 아는 사람 않습니다 내가 어떻게 모든 항목에 그것을 할 수 있을까?

Answer 1

첫째, 당신은 어떻게 당신의 라인 (getValueForKeyAs[T])의 지정된 키에 대한 값을 추출하는 기능이 필요합니다

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

이 당신에게 유형의 RDD를 제공해야합니다 RDD[(Float,Float)]

getValueForKeyAs 수

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)을 포함하는 튜플로 변환합니다.

이

// 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 
} 

우리는 그 함수를 사용

한 가지 가능한 방법은 다음과 같이 요소 "나이"와 "평판"의 값을 추출하기 위해 문자열 연산을 사용하여 RDD의 각 요소를 변환하는 것입니다 모든 기록에서 흥미로운 값을 추출합니다 : 우리가이 일을하기 전에 "나이"를 필터링 할 필요가 없습니다 방법

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

참고. "Age"또는 "Reputation"이없는 레코드를 처리하는 경우 findElement은 None을 반환합니다. 이후 for-comprehension의 결과는 None이고 작업으로 기록은 으로 평평하게이됩니다.

이 문제에 접근하는 더 좋은 방법은 구조화 된 XML 데이터를 처리하는 것입니다. 스칼라는 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의 조합을 사용하여 필요한 모든 정보가 포함되지 않은 레코드를 제거합니다.