Как создать и удалить таблицы в Apache Hadoop c использованием PySpark
Сегодня я расскажу, как затратив минимум усилий при работе с большими данными, справиться с задачей создания таблиц с нужными параметрами, а также, как удалить сразу большое количество потерявших актуальность таблиц.
Текущую деятельность уже невозможно представить без использования больших данных. Предлагаю рассмотреть некоторые методы оптимизации своих усилий в работе с большими данными при решении задач создания и удаления таблиц. Но для начала разберемся в терминах.
Apache Hadoop – это пакет утилит, библиотек и фреймворков, его используют для построения систем, которые работают с Big Data. Он позволяет хранить и обрабатывать данные для выгрузки в другие сервисы.
Hive – система хранения данных, которая помогает запрашивать большие наборы данных в HDFS, использует HQL (Hive Query Language) – язык запросов Hive. Для того чтобы использовать данные выборок Apache Hadoop для дальнейшей их обработки и анализа, можно сохранять их в свои пользовательские таблицы. Существует несколько способов создания таблиц. Самый «лёгкий» способ сохранить данные - написать в Hive QL запрос:
Create table database_lab.mytable as Select a.* from databasename.tablename a;
Здесь и далее указан «простой select», но в вашем случае это может быть сложный запрос с большим количеством таблиц (с использованием join). Выбираем таблицу, структуру которой хотим воссоздать. Данный способ нежелателен, так как таблица будет создаваться в процессе запроса. Другой способ – создать совсем новую таблицу:
Create table database_lab.mytable (column1 string, column2 bigint,……. , columnN int);
После этого уже можно будет заливать выбранные данные в готовую таблицу:
Insert into database_lab.mytable Select a.* from databasename.tablename a;
В данном методе самой трудоёмкой задачей является описание полей и их тип данных. Для решения этой задачи я предлагаю использовать PySpark. Запускаю пользовательскую сессию PySpark с помощью Jupyter NoteBook. Устанавливаю конфигурацию параметров SparkContext и подключаюсь к Apache Hadoop c использованием PySpark:
import os
import sys
from PySpark import SparkConf
from PySpark.sql import HiveContext, SQLContext
conf = SparkConf().setAppName('Create Drop Table').setMaster('yarn') conf.setAll([ ## здесь ваши настройки ])
spark = SparkSession.builder.config(conf=conf).enableHiveSupport().getOrCreate()
hivecontext = HiveContext(spark)
sqlContext =SQLContext(hivecontext)
Следующий шаг – напишу запрос, с помощью которого получаю данные для анализа:
query = “Select a.* from databasename.tablename a”
Создаю spark DataFrame, эта операция не нагружает систему, так как фактически к данным обращения нет:
df_sql = sqlContext.sql(query)
Набираю следующую команду:
df_sql DataFrame[column1 : string, column2: bigint,……. , columnN : int]
Видим практически готовую структуру для создания таблицы. Сделаю несколько преобразований и получаю готовый скрипт для создания пустой таблицы:
# возьмём это значение и преобразуем в строку df_sq_str = str(df_sql) df_sq_str ‘DataFrame[column1 : string, column2: bigint,……. , columnN : int]’ # заменим значения “DataFrame[, ], :” columns_table = df_sq_str.replace(‘DataFrame[’,’’) .replace(‘]’,’’) .replace(‘:’,’’) # напишем команду запроса для создания таблицы hql_create_table = ‘Create table database_lab.mytable ’ + ‘(‘ + columns_table + ‘)’ hql_create_table ‘Create table database_lab.mytable (column1 string, column2 bigint,……. , columnN int)’
На последнем этапе запускаю готовую команду. Это можно осуществить как в PySpark, так и в Hive.
Рано или поздно наступает момент, когда данные таблиц уже использованы и потеряли свою актуальность. Рассмотрю, как удалить свои таблицы с неактуальными данными. Если у вас не так много таблиц в Hadoop, то проблем с удалением не возникает.
Набираю в Hive QL:
Drop table database_lab.mytable;
Однако, когда их больше 1000, и кроме ваших таблиц есть таблицы других пользователей, то возникает вопрос как быстро удалить уже не нужные вам таблицы. Для этого необходимо при создании таблиц стандартизировать их название. Например:
tmp_IvanovVS_kreditfl_2022_08_08_1
где: tmp – временная (промежуточная),
IvanovVS – идентификатор принадлежности, kreditfl – тип данных, 2022_08_08 – дата создания, 1 – номер таблицы, если их больше 1.
Для удаления таблиц снова воспользуюсь возможностями PySpark. Запускаю пользовательскую сессию PySpark с помощью Jupyter NoteBook, устанавливаю конфигурацию параметров SparkContext и подключаюсь к Apache Hadoop c использованием PySpark как указано выше. Создаю Spark DataFrame таблиц базы данных:
df_table_list = spark.sql(“””show tables in database_lab”””)
#Выберем списком названия таблиц
tables_all = [row.tablename for row in df_table_list]
#создадим список моих таблиц
tables_me = []
for table_name in tables_all:
if ‘IvanovVS’ in table_name:
tables_me.append(table_name)
Теперь можно удалить все таблицы:
for table_name in tables_me:
query =”drop table database_lab.” + table_name
df_drop_table = sqlContext.sql(query)
print(‘Удалена таблица {0}’.format(table_name))
Таким образом будут удалены все ваши таблицы. Можно также выбирать таблицы для удаления с помощью дополнительных параметров, заложенных в названии таблиц.
В результате, используя предложенные методы PySpark, можно значительно упростить работу по созданию и удалению собственных таблиц в Apache Hadoop.
