스파크 SQL 쿼리

Question

내 데이터 세트가 이런 식으로 분할 범위 :

Year=yyyy 
|---Month=mm 
| |---Day=dd 
| | |---<parquet-files> 

두 날짜 사이에 데이터가로드 스파크에서 dataframe를 만들 수있는 가장 쉽고 효율적인 방법은 무엇입니까?

Answer 1

당신이 절대적으로이 분할 전략에 충실해야하는 경우, 대답은 당신이 파티션을 부담 할 의향이 있는지 여부에 따라 달라집니다 발견 비용 또는 아닙니다.

Spark가 새 파일을 추가 할 때까지 한 번만 수행해야하는 모든 파티션을 검색하려는 경우 기본 경로를로드 한 다음 파티션 열을 사용하여 필터링 할 수 있습니다.

수백만 개의 파일이 있기 때문에 Spark에 모든 파티션을 검색하지 않으려면 효율적인 전체 솔루션은 쿼리 할 간격을 여러 하위 간격으로 나누면 쉽게 쿼리 할 수 ​​있습니다 @ r0bb23의 접근 방식을 사용하고 나서 함께 결합한다.

위의 두 가지 경우를 최상으로 유지하고 안정된 스키마가 필요하면 외부 파티션 테이블을 정의하여 메타 스토어에 파티션을 등록 할 수 있습니다. 메타 스토어 관리 테이블이 스키마 진화를 현저하게 관리하므로 스키마를 발전 시키려면이 작업을 수행하지 마십시오. 이 확실히 가능하다에 대한 일반적인 코드 작성

// With full discovery 
spark.read.parquet("hdfs:///basepath") 
    .where('Year === 2017 && (
    ('Month === 10 && 'Day >= 6') || ('Month === 11 && 'Day <= 3') 
)) 

// With partial discovery 
val df1 = spark.read.option("basePath", "hdfs:///basepath/") 
    .parquet("hdfs:///basepath/Year=2017/Month=10/Day={0[6-9], [1-3][0-9]}/*/") 
val df2 = spark.read.option("basePath", "hdfs:///basepath/") 
    .parquet("hdfs:///basepath/Year=2017/Month=11/Day={0[1-3]}/*/") 
val df = df1.union(df2) 

하지만 난 그것을 발생하지 않은 : 예를 들어

, 당신은 할 것 2017-10-06 사이 2017-11-03 쿼리합니다. 더 나은 접근법은 질문에 대한 의견에서 설명한 방식으로 파티션을 나누는 것입니다. 테이블이 /basepath/ts=yyyymmddhhmm/*.parquet 같은 것을 사용하여 분할 된 경우 그 대답은 간단하다 : 그것은 추가 시간 & 분 가치가 왜

spark.read.parquet("hdfs:///basepath") 
    .where('ts >= 201710060000L && 'ts <= 201711030000L) 

이유는 당신이 다음에 관계없이 데이터가 분할 여부의 간격을 처리하는 일반적인 코드를 작성할 수 있다는 것입니다 주, 시간, 또는 15 분마다. 실제로 동일한 테이블에서 서로 다른 세분화 된 데이터를 관리 할 수도 있습니다. 예를 들어 구 데이터를 상위 레벨에서 집계하여 발견해야하는 파티션의 총 수를 줄일 수 있습니다.

Answer 2

주소 설명에 여러로드 경로를 추가하도록 편집되었습니다.

정규식 스타일 구문을 사용할 수 있습니다.

val dataset = spark 
    .read 
    .format("parquet") 
    .option("filterPushdown", "true") 
    .option("basePath", "hdfs:///basepath/") 
    .load("hdfs:///basepath/Year=2017/Month=10/Day={0[6-9],[1-3][0-9]}/*/", 
    "hdfs:///basepath/Year=2017/Month=11/Day={0[1-3]}/*/") 

How to use regex to include/exclude some input files in sc.textFile?

주 : 당신이 필요하지 않은 X=* 모든 일, 달 등

당신은 아마 일부에 대한 책을 읽은을해야 할 경우 당신은 *을 수행 할 수 있습니다 Predicate Pushdown (즉, filterPushdown을 위의 true로 설정).

마지막으로, 위의 basepath 옵션을 알 수 있습니다, 그 이유는 여기에서 찾을 수 있습니다 : Prevent DataFrame.partitionBy() from removing partitioned columns from schema