내 pyspark 코드에서 S3의 Amazon kinesis 스트림 파일에 액세스하는 방법?

Question

'snappy'를 사용하여 압축 된 형식의 Kinesis 스트림 파일로 구성된 버킷이 S3에 제공됩니다. 폴더 구조 및 파일 형식은 s3://<mybucket>/yyyy/mm/dd/*.snappy입니다.

sqlContext에 pyspark으로 이것을 읽으려고합니다. 일반적으로 버킷을 다음과 같이 지정합니다.

df = sqlContext.read.json('s3://<mybucket>/inputfile.json') 

이러한 여러 부분 압축 파일을 데이터 프레임에 어떻게 가져 옵니까?

업데이트 :이 같은 구문을 사용하여 더 많은 발전을 한 것처럼 보입니다.

# 
# java.lang.OutOfMemoryError: Java heap space 
# -XX:OnOutOfMemoryError="kill %p 
kill -9 %p" 
# Executing /bin/sh -c "kill 6128 
kill -9 6128"... 

데이터의 크기가 큰 것이 아니라 어떻게 든이 압축 해제 단계는 설상가상으로 보인다 그러나, 힙 크기 문제로 실행. 처음 세 * 년대는 /yyyy/mm/dd/ 하위 폴더를 참조하는 경우

s3://<mybucket>/*/*/*/*.snappy 

: 모든 일/월/년에서 모든 물어 파일을 얻으려고 노력하는 경우

Answer 1

, 이런 일을하려고합니다.

가 testDirectories 폴더를 생성 ... 그리고 그 안에 어떤 날짜 폴더를 중첩 :

는 다음과 같은 테스트를 수행 할 수 있습니다,이 작품을 증명합니다.

nestedDirectories/ 
-- 2016/ 
-- -- 12/ 
-- -- -- 15/ 
-- -- -- -- data.txt 

과 data.txt 내부

:

hello 
world 
I 
Have 
Some 
Words 

그리고 나는 pyspark을 실행했습니다

>>> rdd = sc.readText("/path/to/nestedDirectories/*/*/*/*.txt") 
>>> rdd.count() 
6 

그래서, 그 별 패턴은 RDD로 파일을 가져 오기 위해 작동합니다.

여기에서 메모리와 물건에 문제가있는 경우 파일 크기가 너무 작은 파일이 너무 많을 수 있습니다. 이를 "작은 파일 문제"라고합니다. https://forums.databricks.com/questions/480/how-do-i-ingest-a-large-number-of-files-from-s3-my.html