Выполнение запросов непосредственно к разделенным файлам с помощью Pyspark

Question

Здесь упоминается, что мы можем выполнить запрос непосредственно к одному файлу, как этот.

df = spark.sql("SELECT * FROM parquet.`examples/src/main/resources/users.parquet`")

Мой вопрос: можем ли мы сделать это для набора файлов, разделенных на гггг / мм / дд? На той же странице говорится, что искра может автоматически обнаружить разделы. Я не уверен, как это сделать для этого типа запроса.

Answer 1

Если у вас есть паркетный стол с вложенным partition fields как yyyy/mm/dd затем укажите каталог местоположения таблицы в вашем запросе, тогда spark сможет прочитать все вложенные разделы и создать фрейм данных .

Пример:

У меня есть таблица с 3 полями строки раздела (год, месяц, дата)

hive> desc i;

+--------------------------+-----------------------+-----------------------+--+
|         col_name         |       data_type       |        comment        |
+--------------------------+-----------------------+-----------------------+--+
| id                       | int                   |                       |
| year                     | string                |                       |
| month                    | string                |                       |
| dd                       | string                |                       |
|                          | NULL                  | NULL                  |
| # Partition Information  | NULL                  | NULL                  |
| # col_name               | data_type             | comment               |
|                          | NULL                  | NULL                  |
| year                     | string                |                       |
| month                    | string                |                       |
| dd                       | string                |                       |
+--------------------------+-----------------------+-----------------------+--+

Теперь я пытаюсь загрузить данные, читая файлы паркета (расположение таблицы /apps/hive/warehouse/i).

Файлы в HDFS:

hadoop fs -ls -R /apps/hive/warehouse/i
drwxrwxrwt   - hive hadoop          0 2018-11-03 00:10 /apps/hive/warehouse/i/year=2018
drwxrwxrwt   - hive hadoop          0 2018-11-03 00:10 /apps/hive/warehouse/i/year=2018/month=10
drwxrwxrwt   - hive hadoop          0 2018-11-03 00:10 /apps/hive/warehouse/i/year=2018/month=10/dd=15
-rwxrwxrwt   3 hive hadoop        214 2018-11-03 00:10 /apps/hive/warehouse/i/year=2018/month=10/dd=15/000000_0

Теперь прочитайте данные из pyspark:

>>> df=spark.sql("select * from parquet.`/apps/hive/warehouse/i`")
>>> df.columns
['id', 'year', 'month', 'dd']
>>> df.show(10,False)
+---+----+-----+---+
|id |year|month|dd |
+---+----+-----+---+
|1  |2018|10   |15 |
+---+----+-----+---+

Если вы хотите читать только файлы определенного года,

>>> df=spark.sql("select * from parquet.`/apps/hive/warehouse/i/year=2018`")
>>> df.columns
['id', 'month', 'dd']
>>> df.show(10,False)
+---+-----+---+
|id |month|dd |
+---+-----+---+
|1  |10   |15 |
+---+-----+---+

Нет столбца для год , потому что мы не читаем данные года , используя pyspark, и мы reading only the month,dd and id fields данные из файлов паркета.

UPDATE:

В случае csv-файлов мы можем следовать similar approach as above, и нам не нужно иметь year,month,day полей, так как spark создаст этих полей при чтении данных в каталоге.

Чтение файлов CSV:

#we are reading the csv files with header  
>>> spark.sql("""CREATE OR REPLACE TEMPORARY view df
             USING csv 
             OPTIONS (header "true", path "/apps/hive/warehouse/i")""")

>>> df=spark.sql("select * from df")
>>> df.show(10,False)
+---+-----+---+
|id |month|dd |
+---+-----+---+
|1  |10   |15 |
+---+-----+---+