スパークSQLクエリ

Question

私のデータセットは、このように分割されている範囲：

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

2つの日付の間のデータをロードスパークでのデータフレームを作成する最も簡単かつ効率的な方法は何ですか？

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