[HADOOP] Spark 및 Python은 사용자 정의 파일 형식 / 생성기를 RDD의 입력으로 사용합니다.

Spark 및 Python은 사용자 정의 파일 형식 / 생성기를 RDD의 입력으로 사용합니다.

Spark에서 입력 가능성에 대해 묻고 싶습니다. http://spark.apache.org/docs/latest/programming-guide.html에서 sc.textFile ()을 사용하여 텍스트 파일을 RDD로 읽을 수 있지만 전처리를하고 싶습니다. 예를 들어 RDD 로의 배포가 이루어지기 전에 내 파일은 JSON 형식 일 수 있습니다. {id : 123, text : "...", value : 6} 추가 처리를 위해 JSON의 특정 필드 만 사용하고 싶습니다.

내 생각은 어떻게 든 파이썬 생성기를 SparkContext의 입력으로 사용할 수 있는지 여부였습니다.

또는 Spark에 더 자연스러운 방법이 있다면 Spark의 파일 만 텍스트가 아닌 사용자 정의를 처리하는 방법이 있습니까?

편집하다:

수락 된 답변이 효과가있는 것 같지만 스파크와 파이썬이 gensim을 사용하여 위키 백과를 구문 분석하려고하는 더 실용적인 다음 질문으로 이동했습니다.

해결법

1. ==============================

   1.가장 빠른 방법은 텍스트 파일을있는 그대로로드하고 결과 RDD에서 원하는 필드를 선택하도록 처리하는 것입니다. 이는 클러스터 전체에서 작업을 병렬화하며 단일 시스템에서 사전 처리를 수행하는 것보다 효율적으로 확장됩니다.

   가장 빠른 방법은 텍스트 파일을있는 그대로로드하고 결과 RDD에서 원하는 필드를 선택하도록 처리하는 것입니다. 이는 클러스터 전체에서 작업을 병렬화하며 단일 시스템에서 사전 처리를 수행하는 것보다 효율적으로 확장됩니다.

   JSON (또는 XML)의 경우 사용자 정의 입력 형식이 필요하지 않다고 생각합니다. PySpark는 Python 환경에서 실행되므로 Python에서 정기적으로 사용 가능한 함수를 사용하여 JSON을 직렬화 해제하고 원하는 필드를 추출 할 수 있습니다.

   예를 들면 다음과 같습니다.

   import json
   
   raw = sc.textFile("/path/to/file.json")
   deserialized = raw.map(lambda x: json.loads(x))
   desired_fields = deserialized.map(lambda x: x['key1'])
   

   desired_fields는 이제 원래 JSON 파일에서 key1 아래의 모든 값에 대한 RDD입니다.

   이 패턴을 사용하여 필드 조합을 추출하거나 공백 등으로 분할 할 수 있습니다.

   desired_fields = deserialized.map(lambda x: (x['key1'] + x['key2']).split(' '))
   

   그리고 이것이 너무 복잡해지면 람다를 원하는 모든 사전 처리를 수행하고 deserialized.map (my_preprocessing_func)를 호출하는 일반 Python 함수로 바꿀 수 있습니다.

2. ==============================

   2.예, SparkContext.parallelize ()를 사용하여 파이썬 변수에서 RDD를 만들 수 있습니다.

   예, SparkContext.parallelize ()를 사용하여 파이썬 변수에서 RDD를 만들 수 있습니다.

   data = [1, 2, 3, 4, 5]
   distData = sc.parallelize(data)
   distData.count()   # 5
   

   이 변수는 반복자 일 수도 있습니다.

from https://stackoverflow.com/questions/26177307/spark-and-python-use-custom-file-format-generator-as-input-for-rdd by cc-by-sa and MIT license